Merge pull request #273 from moov-io/fixup-manual-upload-trigger

pipeline: fixup manual cutoff trigger endpoint

Author: adamdecaf

internal/pipeline/manual_cutoff_times.go

@@ -18,16 +18,21 @@
 package pipeline
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
 
-	"github.com/moov-io/achgateway/internal/service"
 	"github.com/moov-io/base/log"
+	"github.com/moov-io/base/telemetry"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type manuallyTriggeredCutoff struct {
-	C chan error
+	C   chan error
+	ctx context.Context
 }
 
 type manualCutoffBody struct {
@@ -63,16 +68,28 @@ func (fr *FileReceiver) triggerManualCutoff() http.HandlerFunc {
 			return
 		}
 
+		ctx, span := telemetry.StartSpan(r.Context(), "api-manual-cutoff", trace.WithAttributes(
+			attribute.StringSlice("achgateway.shards", body.ShardNames),
+		))
+		defer span.End()
+
 		responses := shardResponses{
 			Shards: make(map[string]*string),
 		}
 
-		for _, xfagg := range fr.shardAggregators {
+		for _, shardName := range body.ShardNames {
+			xfagg, ok := fr.shardAggregators[shardName]
+			if !ok {
+				errString := fmt.Sprintf("unknown shard %s", shardName)
+				responses.Shards[shardName] = &errString
+				continue
+			}
+
 			logger := fr.logger.With(log.Fields{
 				"shard": log.String(xfagg.shard.Name),
 			})
 
-			waiter, err := processManualCutoff(logger, body.ShardNames, xfagg.shard, xfagg)
+			waiter, err := processManualCutoff(ctx, logger, xfagg)
 			if err != nil {
 				errString := err.Error()
 				responses.Shards[xfagg.shard.Name] = &errString
@@ -107,15 +124,12 @@ func (fr *FileReceiver) triggerManualCutoff() http.HandlerFunc {
 	}
 }
 
-func processManualCutoff(logger log.Logger, shardNames []string, shard service.Shard, xfagg *aggregator) (*manuallyTriggeredCutoff, error) {
-	if !exists(shardNames, shard.Name) {
-		return nil, fmt.Errorf("unknown shard %s", shard.Name)
-	}
-
+func processManualCutoff(ctx context.Context, logger log.Logger, xfagg *aggregator) (*manuallyTriggeredCutoff, error) {
 	logger.Info().Log("found shard to manually trigger")
 
 	waiter := manuallyTriggeredCutoff{
-		C: make(chan error, 1),
+		C:   make(chan error, 1),
+		ctx: ctx,
 	}
 	xfagg.cutoffTrigger <- waiter
 	return &waiter, nil
@@ -129,3 +143,11 @@ func exists(names []string, shardName string) bool {
 	}
 	return false
 }
+
+func (fr *FileReceiver) getShardNames() []string {
+	shardNames := make([]string, 0, len(fr.shardAggregators))
+	for _, xfagg := range fr.shardAggregators {
+		shardNames = append(shardNames, xfagg.shard.Name)
+	}
+	return shardNames
+}

internal/pipeline/manual_cutoff_times_test.go

@@ -71,6 +71,32 @@ func TestFileReceiver__ManualCutoff(t *testing.T) {
 	require.Nil(t, resp.Shards["testing"])
 }
 
+func TestFileReceiver__ManualCutoff_RequestedShard_NotConfigured(t *testing.T) {
+	fr, wg := setupFileReceiver(t, nil) // waiter returns 'nil' error
+
+	router := mux.NewRouter()
+	router.Path("/trigger-cutoff").HandlerFunc(fr.triggerManualCutoff())
+
+	var buf bytes.Buffer
+	buf.WriteString(`{"shardNames":["testing", "mordor"]}`)
+
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest("PUT", "/trigger-cutoff", &buf)
+	router.ServeHTTP(w, req)
+
+	wg.Wait()
+	w.Flush()
+	require.Equal(t, http.StatusBadRequest, w.Code)
+
+	var resp shardResponses
+	err := json.NewDecoder(w.Body).Decode(&resp)
+	require.NoError(t, err)
+
+	require.Len(t, resp.Shards, 2)
+	require.Nil(t, resp.Shards["testing"])
+	require.NotNil(t, resp.Shards["mordor"])
+}
+
 func TestFileReceiver__ManualCutoffErr(t *testing.T) {
 	bad := errors.New("bad thing")
 	fr, wg := setupFileReceiver(t, bad) // waiter returns error
@@ -115,6 +141,14 @@ func setupFileReceiver(t *testing.T, waiterResponse error) (*FileReceiver, *sync
 		cutoffTrigger: cutoffTrigger,
 	}
 
+	fr.shardAggregators["ND-live-veridian"] = &aggregator{
+		shard: service.Shard{
+			Name: "ND-live-veridian",
+		},
+		merger:        &MockXferMerging{},
+		cutoffTrigger: cutoffTrigger,
+	}
+
 	var wg sync.WaitGroup
 	wg.Add(1)