Some comments about the code

cmd/playground/playground.go

@@ -61,12 +61,13 @@ func app(ctx context.Context) error {
 
 	doActionsHandler := playground.NewDoActionsHandler(fd.DefaultPluginRegistry, lg)
 
-	mux := http.NewServeMux()
-	mux.Handle("/api/v1/do-actions", doActionsHandler)
+	mux := router(doActionsHandler)
 
 	registry := prometheus.NewRegistry()
 	srvr := defaultServer(ctx, *flagAddr, promhttp.InstrumentMetricHandler(registry, mux))
 
+	lg.Warn("starting to listen", zap.String("addr", *flagAddr), zap.String("debug_addr", *flagDebugAddr))
+
 	go func() {
 		<-ctx.Done()
 		_ = srvr.Close()
@@ -100,3 +101,15 @@ func defaultServer(ctx context.Context, addr string, mux http.Handler) *http.Ser
 		},
 	}
 }
+
+func router(doActions *playground.DoActionsHandler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/api/v1/do-actions":
+			doActions.ServeHTTP(w, r)
+			return
+		default:
+			http.Error(w, "", http.StatusNotFound)
+		}
+	})
+}

playground/handler.go → playground/doactions.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"runtime"
 	"strings"
 	"time"
 
@@ -27,8 +28,7 @@ import (
 var jsoniterConfig = jsoniter.ConfigCompatibleWithStandardLibrary
 
 const (
-	pipelineCapacity   = 1
-	pipelineProcessors = 1
+	pipelineCapacity = 1
 )
 
 type DoActionsRequest struct {
@@ -93,6 +93,7 @@ func (h *DoActionsHandler) doActions(ctx context.Context, req DoActionsRequest)
 		newActions := make([]json.RawMessage, 0, len(req.Actions)*2)
 
 		for _, action := range req.Actions {
+			// trying to get the name of the next action
 			var actionWithType = struct {
 				Type string `json:"type"`
 			}{}
@@ -101,6 +102,7 @@ func (h *DoActionsHandler) doActions(ctx context.Context, req DoActionsRequest)
 			}
 			actionType := jsonEscape(actionWithType.Type)
 
+			// wrap plugin with the debug actions
 			debugActionBefore := json.RawMessage(fmt.Sprintf(`{"type": "debug", "message": "before %s"}`, actionType))
 			debugActionAfter := json.RawMessage(fmt.Sprintf(`{"type": "debug", "message": "after %s"}`, actionType))
 			newActions = append(newActions, debugActionBefore, action, debugActionAfter)
@@ -109,7 +111,7 @@ func (h *DoActionsHandler) doActions(ctx context.Context, req DoActionsRequest)
 		req.Actions = newActions
 	}
 
-	// stdout buffer
+	// pipeline stdout buffer
 	stdoutBuf := new(bytes.Buffer)
 	stdoutBuf.Grow(1 << 10)
 	stdout := preparePipelineLogger(stdoutBuf)
@@ -125,6 +127,7 @@ func (h *DoActionsHandler) doActions(ctx context.Context, req DoActionsRequest)
 		IsStrict:            false,
 	}, metricsRegistry, stdout)
 
+	// callback to collect output events
 	events := make(chan json.RawMessage, len(req.Events))
 	outputCb := func(event *pipeline.Event) {
 		events <- event.Root.EncodeToByte()
@@ -135,9 +138,10 @@ func (h *DoActionsHandler) doActions(ctx context.Context, req DoActionsRequest)
 	}
 
 	p.Start()
-	// push events
+
+	// push events to the pipeline
 	for i, event := range req.Events {
-		p.In(pipeline.SourceID(h.requests.Inc()), "fake", int64(i), event, true)
+		p.In(pipeline.SourceID(h.requests.Inc()), "fake", int64(i+1), event, true)
 	}
 
 	// collect result
@@ -186,7 +190,7 @@ func (h *DoActionsHandler) setupPipeline(p *pipeline.Pipeline, req DoActionsRequ
 	}
 	values := map[string]int{
 		"capacity":   pipelineCapacity,
-		"gomaxprocs": pipelineProcessors,
+		"gomaxprocs": runtime.GOMAXPROCS(0),
 	}
 	if err := fd.SetupActions(p, h.plugins, actionsRaw, values); err != nil {
 		return err

playground/fakeclock.go

@@ -6,16 +6,18 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
+// ZeroClock reports time since "start".
+// Used to print relative time rather than absolute.
 type ZeroClock struct {
 	start time.Time
 }
 
+var _ zapcore.Clock = ZeroClock{}
+
 func NewZeroClock(now time.Time) *ZeroClock {
 	return &ZeroClock{start: now}
 }
 
-var _ zapcore.Clock = ZeroClock{}
-
 func (z ZeroClock) Now() time.Time {
 	diff := time.Since(z.start)
 	return time.Time{}.Add(diff)