shutdownFlow: ignore cancel failing due to workflow missing (#3471)

this can happen when dropping mirror which completed or failed over 30 days ago

Author: serprex

flow/activities/flowable.go

@@ -644,6 +644,12 @@ func (a *FlowableActivity) DropFlowSource(ctx context.Context, req *protos.DropF
 	ctx = context.WithValue(ctx, shared.FlowNameKey, req.FlowJobName)
 	srcConn, err := connectors.GetByNameAs[connectors.CDCPullConnector](ctx, nil, a.CatalogPool, req.PeerName)
 	if err != nil {
+		var notFound *exceptions.NotFoundError
+		if errors.As(err, &notFound) {
+			logger := internal.LoggerFromCtx(ctx)
+			logger.Warn("peer missing, skipping", slog.String("peer", req.PeerName))
+			return nil
+		}
 		return a.Alerter.LogFlowError(ctx, req.FlowJobName,
 			exceptions.NewDropFlowError(fmt.Errorf("[DropFlowSource] failed to get source connector: %w", err)),
 		)
@@ -668,6 +674,12 @@ func (a *FlowableActivity) DropFlowDestination(ctx context.Context, req *protos.
 	ctx = context.WithValue(ctx, shared.FlowNameKey, req.FlowJobName)
 	dstConn, err := connectors.GetByNameAs[connectors.CDCSyncConnector](ctx, nil, a.CatalogPool, req.PeerName)
 	if err != nil {
+		var notFound *exceptions.NotFoundError
+		if errors.As(err, &notFound) {
+			logger := internal.LoggerFromCtx(ctx)
+			logger.Warn("peer missing, skipping", slog.String("peer", req.PeerName))
+			return nil
+		}
 		return a.Alerter.LogFlowError(ctx, req.FlowJobName,
 			exceptions.NewDropFlowError(fmt.Errorf("[DropFlowDestination] failed to get destination connector: %w", err)),
 		)

flow/cmd/handler.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -326,8 +327,8 @@ func (h *FlowRequestHandler) FlowStateChange(
 			"",
 			req.FlowConfigUpdate.GetCdcFlowConfigUpdate(),
 		); err != nil {
-			slog.ErrorContext(ctx, "unable to signal workflow", logs, slog.Any("error", err))
-			return nil, exceptions.NewInternalApiError(fmt.Errorf("unable to signal workflow: %w", err))
+			slog.ErrorContext(ctx, "unable to signal workflow update", logs, slog.Any("error", err))
+			return nil, exceptions.NewInternalApiError(fmt.Errorf("unable to signal workflow update: %w", err))
 		}
 	}
 
@@ -405,10 +406,10 @@ func (h *FlowRequestHandler) handleCancelWorkflow(ctx context.Context, workflowI
 
 	select {
 	case <-errLatch.Chan():
-		if err := errLatch.Wait(); err != nil {
-			slog.ErrorContext(ctx, fmt.Sprintf("unable to cancel PeerFlow workflow: %s. Attempting to terminate.", err.Error()))
-			terminationReason := fmt.Sprintf("workflow %s did not cancel in time.", workflowID)
-			if err := h.temporalClient.TerminateWorkflow(ctx, workflowID, runID, terminationReason); err != nil {
+		if err := errLatch.Wait(); err != nil &&
+			err.Error() != "workflow execution already completed" && !strings.HasPrefix(err.Error(), "workflow not found for ID:") {
+			slog.ErrorContext(ctx, "unable to cancel PeerFlow workflow. Attempting to terminate.", slog.Any("error", err))
+			if err := h.temporalClient.TerminateWorkflow(ctx, workflowID, runID, "workflow did not cancel in time."); err != nil {
 				return fmt.Errorf("unable to terminate PeerFlow workflow: %w", err)
 			}
 		}

flow/e2e/api/api_test.go

@@ -120,15 +120,14 @@ func (s Suite) checkCatalogTableMapping(
 	expectedSourceTableNames []string,
 ) (bool, error) {
 	var configBytes sql.RawBytes
-	err := conn.QueryRow(ctx,
-		"SELECT config_proto FROM flows WHERE name = $1", flowName).Scan(&configBytes)
-	if err != nil {
+	if err := conn.QueryRow(ctx,
+		"SELECT config_proto FROM flows WHERE name = $1", flowName,
+	).Scan(&configBytes); err != nil {
 		return false, err
 	}
 
 	var config protos.FlowConnectionConfigs
-	err = proto.Unmarshal(configBytes, &config)
-	if err != nil {
+	if err := proto.Unmarshal(configBytes, &config); err != nil {
 		return false, err
 	}
 
@@ -1302,3 +1301,22 @@ func (s Suite) TestTableAdditionWithoutInitialLoad() {
 	env.Cancel(s.t.Context())
 	e2e.RequireEnvCanceled(s.t, env)
 }
+
+func (s Suite) TestDropMissing() {
+	conn := s.pg.PostgresConnector.Conn()
+	peer := s.source.GeneratePeer(s.t)
+	var peerId int32
+	require.NoError(s.t, conn.QueryRow(s.t.Context(), "select id from peers where name = $1", peer.Name).Scan(&peerId))
+
+	_, err := conn.Exec(s.t.Context(),
+		"insert into flows (name,source_peer,destination_peer,workflow_id,status) values ('test-drop-missing',$1,$1,'drop-missing-wf-id',$2)",
+		peerId, protos.FlowStatus_STATUS_COMPLETED,
+	)
+	require.NoError(s.t, err)
+
+	_, err = s.FlowStateChange(s.t.Context(), &protos.FlowStateChangeRequest{
+		FlowJobName:        "test-drop-missing",
+		RequestedFlowState: protos.FlowStatus_STATUS_TERMINATING,
+	})
+	require.NoError(s.t, err)
+}

flow/otel_metrics/observables.go

@@ -103,23 +103,26 @@ func (a *Float64SyncGauge) Record(ctx context.Context, value float64, options ..
 }
 
 func buildContextualAttributes(ctx context.Context) metric.MeasurementOption {
-	attributes := make([]attribute.KeyValue, 0)
+	attributes := make([]attribute.KeyValue, 0, 12)
 	flowMetadata := internal.GetFlowMetadata(ctx)
 	if flowMetadata != nil {
+		attributes = append(attributes, attribute.String(FlowNameKey, flowMetadata.FlowName))
+		if flowMetadata.Source != nil {
+			attributes = append(attributes,
+				attribute.Stringer(SourcePeerType, flowMetadata.Source.Type),
+				attribute.String(SourcePeerName, flowMetadata.Source.Name))
+		}
+		if flowMetadata.Destination != nil {
+			attributes = append(attributes,
+				attribute.Stringer(DestinationPeerType, flowMetadata.Destination.Type),
+				attribute.String(DestinationPeerName, flowMetadata.Destination.Name))
+		}
 		attributes = append(attributes,
-			attribute.String(FlowNameKey, flowMetadata.FlowName),
-			attribute.Stringer(SourcePeerType, flowMetadata.Source.Type),
-			attribute.Stringer(DestinationPeerType, flowMetadata.Destination.Type),
-			attribute.String(SourcePeerName, flowMetadata.Source.Name),
-			attribute.String(DestinationPeerName, flowMetadata.Destination.Name),
 			attribute.Stringer(FlowStatusKey, flowMetadata.Status),
-			attribute.Bool(IsFlowResyncKey, flowMetadata.IsResync),
-		)
+			attribute.Bool(IsFlowResyncKey, flowMetadata.IsResync))
 	}
 	additionalMetadata := internal.GetAdditionalMetadata(ctx)
-	attributes = append(attributes,
-		attribute.Stringer(FlowOperationKey, additionalMetadata.Operation),
-	)
+	attributes = append(attributes, attribute.Stringer(FlowOperationKey, additionalMetadata.Operation))
 
 	if activity.IsActivity(ctx) {
 		activityInfo := activity.GetInfo(ctx)