Differentiate high and low-velocity loggers. Add error logging for unary APIs

Author: temporal-nick

proxy/adminservice.go

@@ -24,7 +24,8 @@ type (
 	adminServiceProxyServer struct {
 		adminservice.UnimplementedAdminServiceServer
 		adminClient             adminservice.AdminServiceClient
-		logger                  log.Logger
+		verboseLogger           log.Logger
+		replicationLogger       log.Logger
 		outboundAddressOverride string
 		apiOverrides            *config.APIOverridesConfig
 		metricLabelValues       []string
@@ -43,13 +44,17 @@ func NewAdminServiceProxyServer(
 	reportStreamValue func(idx int32, value int32),
 	logger log.Logger,
 ) adminservice.AdminServiceServer {
-	// The AdminServiceStreams will duplicate the same output for an underlying connection issue hundreds of times.
-	// Limit their output to three times per minute
-	logger = log.NewThrottledLogger(log.With(logger, common.ServiceTag(serviceName)),
+	// Replication streams / APIs will run many hundreds of times per second. Throttle their output
+	// to 3 / min
+	replicationLogger := log.NewThrottledLogger(log.With(logger, common.ServiceTag(serviceName)),
 		func() float64 { return 3.0 / 60.0 })
+	// For config operations, allow most logs so that we can see all the info without putting disk at risk
+	verboseLogger := log.NewThrottledLogger(log.With(logger, common.ServiceTag(serviceName)),
+		func() float64 { return 3.0 })
 	return &adminServiceProxyServer{
 		adminClient:             adminClient,
-		logger:                  logger,
+		verboseLogger:           verboseLogger,
+		replicationLogger:       replicationLogger,
 		outboundAddressOverride: outboundAddressOverride,
 		apiOverrides:            apiOverrides,
 		metricLabelValues:       metricLabelValues,
@@ -58,14 +63,14 @@ func NewAdminServiceProxyServer(
 }
 
 func (s *adminServiceProxyServer) AddOrUpdateRemoteCluster(ctx context.Context, in0 *adminservice.AddOrUpdateRemoteClusterRequest) (*adminservice.AddOrUpdateRemoteClusterResponse, error) {
-	s.logger.Info("Received AddOrUpdateRemoteCluster", tag.Address(in0.FrontendAddress), tag.NewBoolTag("Enabled", in0.GetEnableRemoteClusterConnection()), tag.NewStringsTag("configTags", s.metricLabelValues))
+	s.verboseLogger.Info("Received AddOrUpdateRemoteCluster", tag.Address(in0.FrontendAddress), tag.NewBoolTag("Enabled", in0.GetEnableRemoteClusterConnection()), tag.NewStringsTag("configTags", s.metricLabelValues))
 	if !common.IsRequestTranslationDisabled(ctx) {
 		if len(s.outboundAddressOverride) > 0 {
 			// Override this address so that cross-cluster connections flow through the proxy.
 			// Use a separate "external address" config option because the outbound.listenerAddress may not be routable
 			// from the local temporal server, or the proxy may be deployed behind a load balancer.
 			in0.FrontendAddress = s.outboundAddressOverride
-			s.logger.Info("Overwrote outbound address", tag.Address(in0.FrontendAddress), tag.NewStringsTag("configTags", s.metricLabelValues))
+			s.verboseLogger.Info("Overwrote outbound address", tag.Address(in0.FrontendAddress), tag.NewStringsTag("configTags", s.metricLabelValues))
 		}
 	}
 	return s.adminClient.AddOrUpdateRemoteCluster(ctx, in0)
@@ -92,13 +97,13 @@ func (s *adminServiceProxyServer) DeleteWorkflowExecution(ctx context.Context, i
 }
 
 func (s *adminServiceProxyServer) DescribeCluster(ctx context.Context, in0 *adminservice.DescribeClusterRequest) (*adminservice.DescribeClusterResponse, error) {
-	s.logger.Info("Received DescribeClusterRequest")
+	s.verboseLogger.Info("Received DescribeClusterRequest")
 	resp, err := s.adminClient.DescribeCluster(ctx, in0)
 	if common.IsRequestTranslationDisabled(ctx) {
 		return resp, err
 	}
 	if resp != nil {
-		s.logger.Info("Raw DescribeClusterResponse", tag.NewStringTag("clusterID", resp.ClusterId),
+		s.verboseLogger.Info("Raw DescribeClusterResponse", tag.NewStringTag("clusterID", resp.ClusterId),
 			tag.NewStringTag("clusterName", resp.ClusterName), tag.NewStringTag("version", resp.ServerVersion),
 			tag.NewInt64("failoverVersionIncrement", resp.FailoverVersionIncrement), tag.NewInt64("initialFailoverVersion", resp.InitialFailoverVersion),
 			tag.NewBoolTag("isGlobalNamespaceEnabled", resp.IsGlobalNamespaceEnabled), tag.NewStringsTag("configTags", s.metricLabelValues))
@@ -107,19 +112,19 @@ func (s *adminServiceProxyServer) DescribeCluster(ctx context.Context, in0 *admi
 		responseOverride := s.apiOverrides.AdminSerivce.DescribeCluster.Response
 		if resp != nil && responseOverride.FailoverVersionIncrement != nil {
 			resp.FailoverVersionIncrement = *responseOverride.FailoverVersionIncrement
-			s.logger.Info("Overwrite FailoverVersionIncrement", tag.NewInt64("failoverVersionIncrement", resp.FailoverVersionIncrement),
+			s.verboseLogger.Info("Overwrite FailoverVersionIncrement", tag.NewInt64("failoverVersionIncrement", resp.FailoverVersionIncrement),
 				tag.NewStringsTag("configTags", s.metricLabelValues))
 		}
 	}
 
 	if resp != nil {
-		s.logger.Info("Translated DescribeClusterResponse", tag.NewStringTag("clusterID", resp.ClusterId),
+		s.verboseLogger.Info("Translated DescribeClusterResponse", tag.NewStringTag("clusterID", resp.ClusterId),
 			tag.NewStringTag("clusterName", resp.ClusterName), tag.NewStringTag("version", resp.ServerVersion),
 			tag.NewInt64("failoverVersionIncrement", resp.FailoverVersionIncrement), tag.NewInt64("initialFailoverVersion", resp.InitialFailoverVersion),
 			tag.NewBoolTag("isGlobalNamespaceEnabled", resp.IsGlobalNamespaceEnabled), tag.NewStringsTag("configTags", s.metricLabelValues))
 	}
 	if err != nil {
-		s.logger.Info("Error from remote server!", tag.Error(err), tag.NewStringsTag("configTags", s.metricLabelValues))
+		s.verboseLogger.Info("Error from remote server!", tag.Error(err), tag.NewStringsTag("configTags", s.metricLabelValues))
 	}
 	return resp, err
 }
@@ -132,8 +137,17 @@ func (s *adminServiceProxyServer) DescribeHistoryHost(ctx context.Context, in0 *
 	return s.adminClient.DescribeHistoryHost(ctx, in0)
 }
 
-func (s *adminServiceProxyServer) DescribeMutableState(ctx context.Context, in0 *adminservice.DescribeMutableStateRequest) (*adminservice.DescribeMutableStateResponse, error) {
-	return s.adminClient.DescribeMutableState(ctx, in0)
+func (s *adminServiceProxyServer) DescribeMutableState(ctx context.Context, in0 *adminservice.DescribeMutableStateRequest) (resp *adminservice.DescribeMutableStateResponse, err error) {
+	resp, err = s.adminClient.DescribeMutableState(ctx, in0)
+	if err != nil {
+		// This is a duplicate of the grpc client metrics, but not everyone has metrics set up
+		s.replicationLogger.Error("Failed to describe workflow",
+			tag.NewStringTag("WorkflowId", in0.GetExecution().GetWorkflowId()),
+			tag.NewStringTag("RunId", in0.GetExecution().GetRunId()),
+			tag.NewStringTag("Namespace", in0.GetNamespace()),
+			tag.Error(err), tag.Operation("DescribeMutableState"))
+	}
+	return
 }
 
 func (s *adminServiceProxyServer) GetDLQMessages(ctx context.Context, in0 *adminservice.GetDLQMessagesRequest) (*adminservice.GetDLQMessagesResponse, error) {
@@ -152,12 +166,23 @@ func (s *adminServiceProxyServer) GetNamespace(ctx context.Context, in0 *adminse
 	return s.adminClient.GetNamespace(ctx, in0)
 }
 
-func (s *adminServiceProxyServer) GetNamespaceReplicationMessages(ctx context.Context, in0 *adminservice.GetNamespaceReplicationMessagesRequest) (*adminservice.GetNamespaceReplicationMessagesResponse, error) {
-	return s.adminClient.GetNamespaceReplicationMessages(ctx, in0)
+func (s *adminServiceProxyServer) GetNamespaceReplicationMessages(ctx context.Context, in0 *adminservice.GetNamespaceReplicationMessagesRequest) (resp *adminservice.GetNamespaceReplicationMessagesResponse, err error) {
+	resp, err = s.adminClient.GetNamespaceReplicationMessages(ctx, in0)
+	if err != nil {
+		// This is a duplicate of the grpc client metrics, but not everyone has metrics set up
+		s.replicationLogger.Error("Failed to get namespace replication messages", tag.NewStringTag("Cluster", in0.GetClusterName()),
+			tag.Error(err), tag.Operation("GetNamespaceReplicationMessages"))
+	}
+	return
 }
 
-func (s *adminServiceProxyServer) GetReplicationMessages(ctx context.Context, in0 *adminservice.GetReplicationMessagesRequest) (*adminservice.GetReplicationMessagesResponse, error) {
-	return s.adminClient.GetReplicationMessages(ctx, in0)
+func (s *adminServiceProxyServer) GetReplicationMessages(ctx context.Context, in0 *adminservice.GetReplicationMessagesRequest) (resp *adminservice.GetReplicationMessagesResponse, err error) {
+	resp, err = s.adminClient.GetReplicationMessages(ctx, in0)
+	if err != nil {
+		s.replicationLogger.Error("Failed to get replication messages", tag.NewStringTag("Cluster", in0.GetClusterName()),
+			tag.Error(err), tag.Operation("GetReplicationMessages"))
+	}
+	return
 }
 
 func (s *adminServiceProxyServer) GetSearchAttributes(ctx context.Context, in0 *adminservice.GetSearchAttributesRequest) (*adminservice.GetSearchAttributesResponse, error) {
@@ -255,7 +280,7 @@ func ClusterShardIDtoString(sd history.ClusterShardID) string {
 func (s *adminServiceProxyServer) StreamWorkflowReplicationMessages(
 	targetStreamServer adminservice.AdminService_StreamWorkflowReplicationMessagesServer,
 ) (retError error) {
-	defer log.CapturePanic(s.logger, &retError)
+	defer log.CapturePanic(s.replicationLogger, &retError)
 
 	targetMetadata, ok := metadata.FromIncomingContext(targetStreamServer.Context())
 	if !ok {
@@ -268,7 +293,7 @@ func (s *adminServiceProxyServer) StreamWorkflowReplicationMessages(
 		return err
 	}
 
-	logger := log.With(s.logger,
+	logger := log.With(s.replicationLogger,
 		tag.NewStringTag("source", ClusterShardIDtoString(sourceClusterShardID)),
 		tag.NewStringTag("target", ClusterShardIDtoString(targetClusterShardID)))