intercept calls to cache from build

Author: grambbledook

cmd/go-cache-plugin/commands.go

@@ -4,7 +4,9 @@
 package main
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"github.com/grafana/go-cache-plugin/lib/otel"
@@ -25,22 +27,24 @@ import (
 )
 
 var flags struct {
-	CacheDir      string        `flag:"cache-dir,default=$GOCACHE_DIR,Local cache directory (required)"`
-	LogFile       string        `flag:"log-file,default=trace.log,File used for logs"`
-	S3Bucket      string        `flag:"bucket,default=$GOCACHE_S3_BUCKET,S3 bucket name (required if no --local flag provided)"`
-	S3Region      string        `flag:"region,default=$GOCACHE_S3_REGION,S3 region"`
-	S3Endpoint    string        `flag:"s3-endpoint-url,default=$GOCACHE_S3_ENDPOINT_URL,S3 custom endpoint URL (if unset, use AWS default)"`
-	S3PathStyle   bool          `flag:"s3-path-style,default=$GOCACHE_S3_PATH_STYLE,S3 path-style URLs (optional)"`
-	LocalCache    bool          `flag:"local,default=false,Runs cache in local mode (no S3)"`
-	KeyPrefix     string        `flag:"prefix,default=$GOCACHE_KEY_PREFIX,S3 key prefix (optional)"`
-	MinUploadSize int64         `flag:"min-upload-size,default=$GOCACHE_MIN_SIZE,Minimum object size to upload to S3 (in bytes)"`
-	Concurrency   int           `flag:"c,default=$GOCACHE_CONCURRENCY,Maximum number of concurrent requests"`
-	S3Concurrency int           `flag:"u,default=$GOCACHE_S3_CONCURRENCY,Maximum concurrency for upload to S3"`
-	PrintMetrics  bool          `flag:"metrics,default=$GOCACHE_METRICS,Print summary metrics to stderr at exit"`
-	Expiration    time.Duration `flag:"expiry,default=$GOCACHE_EXPIRY,Cache expiration period (optional)"`
-	Verbose       bool          `flag:"v,default=$GOCACHE_VERBOSE,Enable verbose logging"`
-	DebugLog      int           `flag:"debug,default=$GOCACHE_DEBUG,Enable detailed per-request debug logging (noisy)"`
-	TracingParams string        `flag:"tracing,default=runId:runAttempt:jobName:stepName,Tracing params"`
+	CacheDir             string        `flag:"cache-dir,default=$GOCACHE_DIR,Local cache directory (required)"`
+	LogFile              string        `flag:"log-file,default=trace.log,File used for logs"`
+	S3Bucket             string        `flag:"bucket,default=$GOCACHE_S3_BUCKET,S3 bucket name (required if no --local flag provided)"`
+	S3Region             string        `flag:"region,default=$GOCACHE_S3_REGION,S3 region"`
+	S3Endpoint           string        `flag:"s3-endpoint-url,default=$GOCACHE_S3_ENDPOINT_URL,S3 custom endpoint URL (if unset, use AWS default)"`
+	S3PathStyle          bool          `flag:"s3-path-style,default=$GOCACHE_S3_PATH_STYLE,S3 path-style URLs (optional)"`
+	LocalCache           bool          `flag:"local-cache,default=false,Runs in no cache mode (no S3)"`
+	KeyPrefix            string        `flag:"prefix,default=$GOCACHE_KEY_PREFIX,S3 key prefix (optional)"`
+	MinUploadSize        int64         `flag:"min-upload-size,default=$GOCACHE_MIN_SIZE,Minimum object size to upload to S3 (in bytes)"`
+	Concurrency          int           `flag:"c,default=$GOCACHE_CONCURRENCY,Maximum number of concurrent requests"`
+	S3Concurrency        int           `flag:"u,default=$GOCACHE_S3_CONCURRENCY,Maximum concurrency for upload to S3"`
+	PrintMetrics         bool          `flag:"metrics,default=$GOCACHE_METRICS,Print summary metrics to stderr at exit"`
+	Expiration           time.Duration `flag:"expiry,default=$GOCACHE_EXPIRY,Cache expiration period (optional)"`
+	Verbose              bool          `flag:"v,default=$GOCACHE_VERBOSE,Enable verbose logging"`
+	DebugLog             int           `flag:"debug,default=$GOCACHE_DEBUG,Enable detailed per-request debug logging (noisy)"`
+	TracingEnabled       bool          `flag:"tracing,default=false,Enable tracing (optional)"`
+	TracingContext       string        `flag:"context,default=runId:runAttempt:jobName:stepName:stepNumber,Tracing params"`
+	OtelCollectorAddress string        `flag:"otel-collector,default,OTEL collector address (optional)"`
 }
 
 const (
@@ -66,11 +70,12 @@ func runDirect(env *command.Env) error {
 }
 
 var serveFlags struct {
-	Plugin   string `flag:"plugin,default=$GOCACHE_PLUGIN,Plugin service addr (or port) (required)"`
-	HTTP     string `flag:"http,default=$GOCACHE_HTTP,HTTP service address ([host]:port)"`
-	ModProxy bool   `flag:"modproxy,default=$GOCACHE_MODPROXY,Enable a Go module proxy (requires --http)"`
-	RevProxy string `flag:"revproxy,default=$GOCACHE_REVPROXY,Reverse proxy these hosts (comma-separated; requires --http)"`
-	SumDB    string `flag:"sumdb,default=$GOCACHE_SUMDB,SumDB servers to proxy for (comma-separated)"`
+	Plugin     string `flag:"plugin,default=$GOCACHE_PLUGIN,Plugin service addr (or port) (required)"`
+	HTTP       string `flag:"http,default=$GOCACHE_HTTP,HTTP service address ([host]:port)"`
+	ModProxy   bool   `flag:"modproxy,default=$GOCACHE_MODPROXY,Enable a Go module proxy (requires --http)"`
+	ModNoCache bool   `flag:"mod-nocache,default=false,Disable the local module cache (requires --modproxy)"`
+	RevProxy   string `flag:"revproxy,default=$GOCACHE_REVPROXY,Reverse proxy these hosts (comma-separated; requires --http)"`
+	SumDB      string `flag:"sumdb,default=$GOCACHE_SUMDB,SumDB servers to proxy for (comma-separated)"`
 }
 
 var logger *log.Logger
@@ -83,7 +88,7 @@ func runServe(env *command.Env) error {
 		return env.Usagef("you must provide a --plugin addr (or port)")
 	}
 
-	otel.Init(context.Background(), otel.Config{Mode: otel.ModeStdout, LogFile: flags.LogFile})
+	log.Printf("Otel exporter initialized with address: %s", flags.OtelCollectorAddress)
 	// Initialize the cache server. Unlike a direct server, only close down and
 	// wait for cache cleanup when the whole process exits.
 	s, s3c, err := initCacheServer(env)
@@ -176,8 +181,11 @@ func runServe(env *command.Env) error {
 
 // runConnect implements a direct cache proxy by connecting to a remote server.
 func runConnect(env *command.Env, plugin string) error {
-	addr := plugin
 
+	ctx := env.Context()
+	shutdownTracer, reportSpan, err := initTracing(ctx)
+
+	addr := plugin
 	// If the caller has not specified a host/port, then likely this is an older usage which only specifies port
 	if !strings.Contains(plugin, ":") {
 		port, err := strconv.Atoi(plugin)
@@ -197,23 +205,61 @@ func runConnect(env *command.Env, plugin string) error {
 
 	out := taskgroup.Go(func() error {
 		defer conn.(*net.TCPConn).CloseWrite() // let the server finish
-		return copy(conn, os.Stdin)
+		return copy(conn, os.Stdin, reportSpan)
 	})
-	if rerr := copy(os.Stdout, conn); rerr != nil {
+	if rerr := copy(os.Stdout, conn, reportSpan); rerr != nil {
 		vprintf("read responses: %v", err)
 	}
 	out.Wait()
 	conn.Close()
-	vprintf("connection closed (%v elapsed)", time.Since(start))
+
+	shutdownTracer(context.Background())
+	println("@@@@@ - connection closed @@@@@")
+	println(fmt.Sprintf("connection closed (%v elapsed)", time.Since(start)))
 	return nil
 }
 
+func initTracing(ctx context.Context) (func(context.Context) error, func([]byte), error) {
+	if !flags.TracingEnabled {
+		return func(context.Context) error { return nil }, func([]byte) {}, nil
+	}
+
+	var shutdown func(context.Context) error
+	var err error
+	if flags.OtelCollectorAddress != "" {
+		shutdown, err = otel.SetupOtelTraceProvider(ctx, flags.OtelCollectorAddress)
+	} else if flags.LogFile != "" {
+		log.Printf("Otel Collector address not specified, starting with the logging reporter, log file: %s", flags.LogFile)
+		shutdown, err = otel.SetupLoggingProvider(ctx, flags.LogFile)
+	} else {
+		log.Printf("please specify either --otel-collector or --log-file to setup tracing or disable tracing")
+		return nil, nil, errors.New("otel exporter not initialized")
+	}
+	if err != nil {
+		return nil, nil, err
+	}
+
+	tracingContext := otel.NewTracedFromString(flags.TracingContext)
+
+	spanner := otel.NewAwesomeSpanner(tracingContext)
+
+	spanReporter := func(buffer []byte) {
+		id, err := parseId(buffer[:])
+		if err != nil {
+			log.Printf("failed to parse id from buffer: %v", err)
+		}
+		spanner.ProcessId(ctx, id)
+	}
+
+	return shutdown, spanReporter, err
+}
+
 // copy emulates the base case of io.Copy, but does not attempt to use the
 // io.ReaderFrom or io.WriterTo implementations.
 //
 // TODO(creachadair): For some reason io.Copy does not work correctly when r is
 // a pipe (e.g., stdin) and w is a TCP socket. Figure out why.
-func copy(w io.Writer, r io.Reader) error {
+func copy(w io.Writer, r io.Reader, reportTrace func([]byte)) error {
 	var buf [4096]byte
 	for {
 		nr, err := r.Read(buf[:])
@@ -223,6 +269,8 @@ func copy(w io.Writer, r io.Reader) error {
 			} else if nw < nr {
 				return fmt.Errorf("wrote %d < %d bytes: %w", nw, nr, io.ErrShortWrite)
 			}
+
+			reportTrace(buf[:])
 		}
 		if err == io.EOF {
 			return nil
@@ -231,3 +279,21 @@ func copy(w io.Writer, r io.Reader) error {
 		}
 	}
 }
+
+func parseId(buf []byte) (string, error) {
+	map1 := make(map[string]any)
+
+	err := json.Unmarshal(buf, &map1)
+	if err != nil {
+		slice, _, found := bytes.Cut(buf, []byte{'\n'})
+		if found {
+			err2 := json.Unmarshal(slice, &map1)
+			if err2 != nil {
+				return "", err
+			}
+		}
+	}
+
+	value := map1["ID"]
+	return fmt.Sprintf("map1 = %v", value), nil
+}

cmd/go-cache-plugin/go-cache-plugin.go

@@ -72,7 +72,6 @@ When --http is enabled, the following options are available:
 
 This mode bridges stdin/stdout to a cache server (see the "serve" command)
 listening on the specified port.`,
-
 				Run: command.Adapt(runConnect),
 			},
 			command.HelpCommand(helpTopics),

cmd/go-cache-plugin/setup.go

@@ -11,9 +11,7 @@ import (
 	"errors"
 	"expvar"
 	"fmt"
-	"github.com/grafana/go-cache-plugin/lib/otel"
-	"go.opentelemetry.io/auto/sdk"
-	"go.opentelemetry.io/otel/sdk/trace"
+	"log"
 	"net/http"
 	"os"
 	"path"
@@ -43,31 +41,24 @@ func initCacheServer(env *command.Env) (*gocache.Server, *s3util.Client, error)
 	case flags.CacheDir == "":
 		return nil, nil, env.Usagef("you must provide a --cache-dir")
 	case flags.LocalCache:
-		dir, err := cachedir.New(flags.CacheDir)
+		dirCache, err := cachedir.New(flags.CacheDir)
 		if err != nil {
 			return nil, nil, fmt.Errorf("create local cache: %w", err)
 		}
 
-		sdk.TracerProvider()
-		trace.NewTracerProvider()
-		cache := &gobuild.LocalCache{
-			Local:  dir,
-			Tracer: otel.NewTracedFromString(flags.TracingParams),
-		}
-
-		close := cache.Close
+		cacheClose := func(context.Context) error { return nil }
 		if flags.Expiration > 0 {
-			dirClose := dir.Cleanup(flags.Expiration)
-			close = func(ctx context.Context) error {
-				return errors.Join(cache.Close(ctx), dirClose(ctx))
+			dirClose := dirCache.Cleanup(flags.Expiration)
+			cacheClose = func(ctx context.Context) error {
+				return errors.Join(dirClose(ctx))
 			}
 		}
 
 		setMetrics := func(ctx context.Context, m *expvar.Map) {}
 		s := &gocache.Server{
-			Get:         cache.Get,
-			Put:         cache.Put,
-			Close:       close,
+			Get:         dirCache.Get,
+			Put:         dirCache.Put,
+			Close:       cacheClose,
 			SetMetrics:  setMetrics,
 			MaxRequests: flags.Concurrency,
 			Logf:        vprintf,
@@ -76,7 +67,7 @@ func initCacheServer(env *command.Env) (*gocache.Server, *s3util.Client, error)
 		return s, nil, nil
 
 	case flags.S3Bucket == "" && !flags.LocalCache:
-		return nil, nil, env.Usagef("you must provide an S3 --bucket name")
+		return nil, nil, env.Usagef("you must provide an S3 --bucket name or run with --no-cache")
 	}
 	region, err := getBucketRegion(env.Context(), flags.S3Bucket)
 	if err != nil {
@@ -148,14 +139,18 @@ func initModProxy(env *command.Env, s3c *s3util.Client) (_ http.Handler, cleanup
 		return nil, nil, env.Usagef("you must set --http to enable --modproxy")
 	}
 
-	modCachePath := filepath.Join(flags.CacheDir, "module")
-	if err := os.MkdirAll(modCachePath, 0755); err != nil {
-		return nil, nil, fmt.Errorf("create module cache: %w", err)
+	if s3c == nil && !flags.LocalCache {
+		return nil, nil, errors.New("s3 client not configured")
 	}
 
 	var cacher goproxy.Cacher = nil
-	var metrics func() *expvar.Map
+	var metrics = func() *expvar.Map { return &expvar.Map{} }
 	if s3c != nil {
+		modCachePath := filepath.Join(flags.CacheDir, "module")
+		if err := os.MkdirAll(modCachePath, 0755); err != nil {
+			return nil, nil, fmt.Errorf("create module cache: %w", err)
+		}
+
 		cacher := &modproxy.S3Cacher{
 			Local:       modCachePath,
 			S3Client:    s3c,
@@ -166,25 +161,16 @@ func initModProxy(env *command.Env, s3c *s3util.Client) (_ http.Handler, cleanup
 		}
 		cleanup = func() { vprintf("close cacher (err=%v)", cacher.Close()) }
 		metrics = cacher.Metrics
-	} else {
-		//cache := modproxy.NewLocalModCacher(modCachePath, logger)
-		//cache := modproxy.NewNoopModCacher(logger)
-		//metrics = cache.Metrics
-		//cacher = cache
-		//metrics = func() { return &expvar.Map{} }
-		metrics = func() *expvar.Map { return &expvar.Map{} }
 	}
+
 	proxy := &goproxy.Goproxy{
-		Fetcher: &modproxy.LoggingGoFetcher{
-			Delegate: &goproxy.GoFetcher{
-				// As configured, the fetcher should never shell out to the go
-				// tool. Specifically, because we set GOPROXY and do not set any
-				// bypass via GONOPROXY, GOPRIVATE, etc., we will only attempt to
-				// proxy for the specific server(s) listed in Env.
-				GoBin: "/bin/false",
-				Env:   []string{"GOPROXY=https://proxy.golang.org"},
-			},
-			Tracer: otel.NewTracedFromString(flags.TracingParams),
+		Fetcher: &goproxy.GoFetcher{
+			// As configured, the fetcher should never shell out to the go
+			// tool. Specifically, because we set GOPROXY and do not set any
+			// bypass via GONOPROXY, GOPRIVATE, etc., we will only attempt to
+			// proxy for the specific server(s) listed in Env.
+			GoBin: "/bin/false",
+			Env:   []string{"GOPROXY=https://proxy.golang.org"},
 		},
 		Cacher:        cacher,
 		ProxiedSumDBs: []string{"sum.golang.org"}, // default, see below
@@ -341,6 +327,7 @@ func makeHandler(modProxy, revProxy http.Handler) http.HandlerFunc {
 			return
 		}
 		if modProxy != nil && r.Method == http.MethodGet && strings.HasPrefix(path, "/mod/") {
+			log.Printf("proxying %s %s", r.Method, r.URL.Path)
 			modProxy.ServeHTTP(w, r)
 			return
 		}

lib/gobuild/noop.go

@@ -2,9 +2,6 @@ package gobuild
 
 import (
 	"github.com/creachadair/gocache"
-	"github.com/creachadair/gocache/cachedir"
-	"github.com/grafana/go-cache-plugin/lib/otel"
-	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/net/context"
 )
 
@@ -14,28 +11,17 @@ type TheCache interface {
 	Close(ctx context.Context) error
 }
 
-type LocalCache struct {
-	Local  *cachedir.Dir
-	Tracer *otel.TraceMeta
+type NoopCache struct {
 }
 
-func (l *LocalCache) Get(ctx context.Context, actionID string) (outputID, diskPath string, _ error) {
-	_, span := l.Tracer.SpanWithContext(ctx, "BUILD-GET", attribute.KeyValue{Key: "action_id", Value: attribute.StringValue(actionID)})
-	defer func() {
-		span.End()
-	}()
-
-	return l.Local.Get(ctx, actionID)
+func (l *NoopCache) Get(ctx context.Context, actionID string) (outputID, diskPath string, _ error) {
+	return "", "", nil
 }
 
-func (l *LocalCache) Put(ctx context.Context, obj gocache.Object) (diskPath string, _ error) {
-	_, span := l.Tracer.SpanWithContext(ctx, "BUILD-PUT", attribute.KeyValue{Key: "action_id", Value: attribute.StringValue(obj.ActionID)})
-	defer func() {
-		span.End()
-	}()
-	return l.Local.Put(ctx, obj)
+func (l *NoopCache) Put(ctx context.Context, obj gocache.Object) (diskPath string, _ error) {
+	return "", nil
 }
 
-func (l *LocalCache) Close(ctx context.Context) error {
+func (l *NoopCache) Close(ctx context.Context) error {
 	return nil
 }

lib/modproxy/noop.go

@@ -25,14 +25,14 @@ func (n *NoopCacher) Put(ctx context.Context, name string, content io.ReadSeeker
 
 type LocalCache struct {
 	Local  goproxy.Cacher
-	Tracer *otel.TraceMeta
+	Tracer *otel.TracingContext
 }
 
-func NewLocalModCacher(path string, tracer *otel.TraceMeta) *LocalCache {
+func NewLocalModCacher(path string, tracer *otel.TracingContext) *LocalCache {
 	return &LocalCache{Local: goproxy.DirCacher(path), Tracer: tracer}
 }
 
-func NewNoopModCacher(tracer *otel.TraceMeta) *LocalCache {
+func NewNoopModCacher(tracer *otel.TracingContext) *LocalCache {
 	return &LocalCache{Local: &NoopCacher{}, Tracer: tracer}
 }
 
@@ -59,7 +59,7 @@ func (l *LocalCache) Metrics() *expvar.Map {
 
 type LoggingGoFetcher struct {
 	Delegate goproxy.Fetcher
-	Tracer   *otel.TraceMeta
+	Tracer   *otel.TracingContext
 }
 
 func (l *LoggingGoFetcher) Query(ctx context.Context, path, query string) (string, time.Time, error) {