chore: ignore context canceled errors when publishing to pubsub (#1185)

Signed-off-by: Alessandro Yuichi Okimoto <[email protected]>
bucketeer-io · Aug 6, 2024 · 14454f3 · 14454f3
1 parent 63ab2f7
commit 14454f3
Show file tree

Hide file tree

Showing 2 changed files with 63 additions and 65 deletions.
diff --git a/pkg/autoops/api/flag_trigger_operation.go b/pkg/autoops/api/flag_trigger_operation.go
@@ -23,7 +23,6 @@ import (
  ftdomain "github.com/bucketeer-io/bucketeer/pkg/feature/domain"
  ftstorage "github.com/bucketeer-io/bucketeer/pkg/feature/storage/v2"
  "github.com/bucketeer-io/bucketeer/pkg/locale"
- "github.com/bucketeer-io/bucketeer/pkg/log"
  autoopsproto "github.com/bucketeer-io/bucketeer/proto/autoops"
 )
 
@@ -63,14 +62,7 @@ func enableFeature(
  logger *zap.Logger,
 ) error {
  if err := feature.Enable(); err != nil {
- logger.Error(
- "Failed to enable feature flag",
- log.FieldsFromImcomingContext(ctx).AddFields(
- zap.Error(err),
- zap.String("featureId", feature.Id),
- zap.String("environmentNamespace", environmentNamespace),
- )...,
- )
+ // If the flag is already disabled, we skip the updating
  return nil
  }
  if err := ftStorage.UpdateFeature(ctx, feature, environmentNamespace); err != nil {
@@ -90,14 +82,7 @@ func disableFeature(
  logger *zap.Logger,
 ) error {
  if err := feature.Disable(); err != nil {
- logger.Error(
- "Failed to disable feature flag",
- log.FieldsFromImcomingContext(ctx).AddFields(
- zap.Error(err),
- zap.String("featureId", feature.Id),
- zap.String("environmentNamespace", environmentNamespace),
- )...,
- )
+ // If the flag is already disabled, we skip the updating
  return nil
  }
  if err := ftStorage.UpdateFeature(ctx, feature, environmentNamespace); err != nil {

diff --git a/pkg/gateway/api/api_grpc.go b/pkg/gateway/api/api_grpc.go
@@ -16,6 +16,7 @@ package api
 
 import (
  "context"
+ "errors"
  "time"
 
  "github.com/golang/protobuf/ptypes"
@@ -309,15 +310,17 @@ func (s *grpcGatewayService) GetEvaluations(
  defer span.End()
  envAPIKey, err := s.checkRequest(ctx, []accountproto.APIKey_Role{accountproto.APIKey_SDK_CLIENT})
  if err != nil {
- s.logger.Error("Failed to check GetEvaluations request",
- log.FieldsFromImcomingContext(ctx).AddFields(
- zap.Error(err),
- zap.String("tag", req.Tag),
- zap.Any("user", req.User),
- zap.Any("sourceId", req.SourceId),
- zap.String("sdkVersion", req.SdkVersion),
- )...,
- )
+ if !errors.Is(err, context.Canceled) {
+ s.logger.Error("Failed to check GetEvaluations request",
+ log.FieldsFromImcomingContext(ctx).AddFields(
+ zap.Error(err),
+ zap.String("tag", req.Tag),
+ zap.Any("user", req.User),
+ zap.Any("sourceId", req.SourceId),
+ zap.String("sdkVersion", req.SdkVersion),
+ )...,
+ )
+ }
  return nil, err
  }
  projectID := envAPIKey.ProjectId
@@ -500,16 +503,18 @@ func (s *grpcGatewayService) GetEvaluation(
  defer span.End()
  envAPIKey, err := s.checkRequest(ctx, []accountproto.APIKey_Role{accountproto.APIKey_SDK_CLIENT})
  if err != nil {
- s.logger.Error("Failed to check GetEvaluation request",
- log.FieldsFromImcomingContext(ctx).AddFields(
- zap.Error(err),
- zap.String("tag", req.Tag),
- zap.Any("user", req.User),
- zap.String("featureId", req.FeatureId),
- zap.Any("sourceId", req.SourceId),
- zap.String("sdkVersion", req.SdkVersion),
- )...,
- )
+ if !errors.Is(err, context.Canceled) {
+ s.logger.Error("Failed to check GetEvaluation request",
+ log.FieldsFromImcomingContext(ctx).AddFields(
+ zap.Error(err),
+ zap.String("tag", req.Tag),
+ zap.Any("user", req.User),
+ zap.String("featureId", req.FeatureId),
+ zap.Any("sourceId", req.SourceId),
+ zap.String("sdkVersion", req.SdkVersion),
+ )...,
+ )
+ }
  return nil, err
  }
  requestTotal.WithLabelValues(
@@ -611,15 +616,17 @@ func (s *grpcGatewayService) GetFeatureFlags(
  defer span.End()
  envAPIKey, err := s.checkRequest(ctx, []accountproto.APIKey_Role{accountproto.APIKey_SDK_SERVER})
  if err != nil {
- s.logger.Error("Failed to check GetFeatureFlags request",
- log.FieldsFromImcomingContext(ctx).AddFields(
- zap.Error(err),
- zap.String("tag", req.Tag),
- zap.String("featureFlagsId", req.FeatureFlagsId),
- zap.Any("sourceId", req.SourceId),
- zap.String("sdkVersion", req.SdkVersion),
- )...,
- )
+ if !errors.Is(err, context.Canceled) {
+ s.logger.Error("Failed to check GetFeatureFlags request",
+ log.FieldsFromImcomingContext(ctx).AddFields(
+ zap.Error(err),
+ zap.String("tag", req.Tag),
+ zap.String("featureFlagsId", req.FeatureFlagsId),
+ zap.Any("sourceId", req.SourceId),
+ zap.String("sdkVersion", req.SdkVersion),
+ )...,
+ )
+ }
  return nil, err
  }
  projectID := envAPIKey.ProjectId
@@ -769,14 +776,16 @@ func (s *grpcGatewayService) GetSegmentUsers(
  defer span.End()
  envAPIKey, err := s.checkRequest(ctx, []accountproto.APIKey_Role{accountproto.APIKey_SDK_SERVER})
  if err != nil {
- s.logger.Error("Failed to check GetSegmentUsers request",
- log.FieldsFromImcomingContext(ctx).AddFields(
- zap.Error(err),
- zap.Strings("segmentIds", req.SegmentIds),
- zap.Any("sourceId", req.SourceId),
- zap.String("sdkVersion", req.SdkVersion),
- )...,
- )
+ if !errors.Is(err, context.Canceled) {
+ s.logger.Error("Failed to check GetSegmentUsers request",
+ log.FieldsFromImcomingContext(ctx).AddFields(
+ zap.Error(err),
+ zap.Strings("segmentIds", req.SegmentIds),
+ zap.Any("sourceId", req.SourceId),
+ zap.String("sdkVersion", req.SdkVersion),
+ )...,
+ )
+ }
  return nil, err
  }
  projectID := envAPIKey.ProjectId
@@ -1408,32 +1417,35 @@ func (s *grpcGatewayService) RegisterEvents(
  typ string,
  ) map[string]*gwproto.RegisterEventsResponse_Error {
  errs := make(map[string]*gwproto.RegisterEventsResponse_Error)
- errors := p.PublishMulti(ctx, messages)
+ multiErrs := p.PublishMulti(ctx, messages)
  var repeatableErrors, nonRepeateableErrors float64
- for id, err := range errors {
+ for id, err := range multiErrs {
  retriable := err != publisher.ErrBadMessage
  if retriable {
  repeatableErrors++
  } else {
  nonRepeateableErrors++
  }
- s.logger.Error(
- "Failed to publish event",
- log.FieldsFromImcomingContext(ctx).AddFields(
- zap.Error(err),
- zap.String("environmentID", envAPIKey.Environment.Id),
- zap.String("eventType", typ),
- zap.String("id", id),
- )...,
- )
+ if !errors.Is(err, context.Canceled) {
+ s.logger.Error(
+ "Failed to publish event",
+ log.FieldsFromImcomingContext(ctx).AddFields(
+ zap.Error(err),
+ zap.String("environmentID", envAPIKey.Environment.Id),
+ zap.String("eventType", typ),
+ zap.String("id", id),
+ zap.Any("sourceId", req.SourceId),
+ )...,
+ )
+ }
  errs[id] = &gwproto.RegisterEventsResponse_Error{
  Retriable: retriable,
  Message: "Failed to publish event",
  }
  }
  eventCounter.WithLabelValues(callerGatewayService, typ, codeNonRepeatableError).Add(nonRepeateableErrors)
  eventCounter.WithLabelValues(callerGatewayService, typ, codeRepeatableError).Add(repeatableErrors)
- eventCounter.WithLabelValues(callerGatewayService, typ, codeOK).Add(float64(len(messages) - len(errors)))
+ eventCounter.WithLabelValues(callerGatewayService, typ, codeOK).Add(float64(len(messages) - len(multiErrs)))
  return errs
  }
  for i, event := range req.Events {
@@ -1454,6 +1466,7 @@ func (s *grpcGatewayService) RegisterEvents(
  zap.String("eventID", event.Id),
  zap.String("environmentNamespace", event.EnvironmentNamespace),
  zap.Any("event", event.Event),
+ zap.Any("sourceId", req.SourceId),
  )
  continue
  }