Grpc opts

libs/capability/config.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/loop"
 	"github.com/smartcontractkit/chainlink-common/x/config/cli"
 	"github.com/smartcontractkit/chainlink-common/x/config/markup/tomlmarkup"
 )
@@ -19,11 +20,13 @@ const settingsNamespace = "settings"
 
 const defaultHost = "localhost"
 
-// GRPCConfig configures the gRPC servers this process opens.
+// GRPCConfig configures the gRPC clients and servers this process opens.
 type GRPCConfig struct {
 	// AdvertiseHost is the host the gRPC servers this process opens bind to and are advertised at.
 	// Empty binds every interface.
 	AdvertiseHost string
+
+	loop.GRPCOptsConfig
 }
 
 // the standard configuration for a capability; does not include the config that a capability may request via its constructor.

libs/capability/embed.go

@@ -7,6 +7,7 @@ import (
 	"reflect"
 
 	"github.com/jonboulle/clockwork"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/samber/do/v2"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry"
 
@@ -39,6 +40,7 @@ type InstanceInfo struct {
 type startConfig struct {
 	capabilities CapabilitiesConfig
 	grpc         GRPCConfig
+	grpcOpts     *loop.GRPCOpts
 	settings     *loop.AtomicSettings
 	overrides    []func(*Injector) error
 }
@@ -52,10 +54,11 @@ func WithProvider(provide func(*Injector) error) StartOption {
 	return func(c *startConfig) { c.overrides = append(c.overrides, provide) }
 }
 
-func withNode(capabilities CapabilitiesConfig, grpc GRPCConfig, settings *loop.AtomicSettings) StartOption {
+func withNode(capabilities CapabilitiesConfig, grpc GRPCConfig, grpcOpts *loop.GRPCOpts, settings *loop.AtomicSettings) StartOption {
 	return func(c *startConfig) {
 		c.capabilities = capabilities
 		c.grpc = grpc
+		c.grpcOpts = grpcOpts
 		c.settings = settings
 	}
 }
@@ -73,6 +76,13 @@ func StartCapabilities[T Capability](ctx context.Context, lggr logger.Logger, ba
 	for _, opt := range opts {
 		opt(cfg)
 	}
+	if cfg.grpcOpts == nil {
+		// Only run() gives gRPC options. Other callers, for example capabilitytest, come here.
+		// Tests call this function many times in one process. Each call must use a new registry,
+		// because the default registry panics if you register the same metrics two times.
+		grpcOpts := loop.GRPCOptsConfig{Registerer: prometheus.NewRegistry()}.New(lggr)
+		cfg.grpcOpts = &grpcOpts
+	}
 
 	var built []T
 	var svcs []services.Service
@@ -82,7 +92,7 @@ func StartCapabilities[T Capability](ctx context.Context, lggr logger.Logger, ba
 			instLggr = logger.Named(lggr, fmt.Sprintf("Instance%d", id))
 		}
 
-		reg, err := startRegistry(ctx, instLggr, cfg.capabilities, cfg.grpc)
+		reg, err := startRegistry(ctx, instLggr, cfg.capabilities, cfg.grpc, *cfg.grpcOpts)
 		if err != nil {
 			return built, svcs, fmt.Errorf("failed to start registry: %w", err)
 		}

libs/capability/grpc_compat_test.go

@@ -0,0 +1,27 @@
+package capability
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/loop"
+)
+
+// TestGRPCServerMaxRecvMsgSizeRoundTripsThroughLoopEnv makes sure that this process reads the same
+// env var as loop. The LOOP host sets the limit with loop.EnvConfig.AsCmdEnv. GRPCConfig must then
+// contain the same limit.
+func TestGRPCServerMaxRecvMsgSizeRoundTripsThroughLoopEnv(t *testing.T) {
+	env := loop.EnvConfig{GRPCServerMaxRecvMsgSize: 7 << 20}
+	for _, kv := range env.AsCmdEnv() {
+		k, v, ok := strings.Cut(kv, "=")
+		require.True(t, ok, "malformed env entry %q", kv)
+		if strings.HasPrefix(k, "CL_GRPC_") {
+			t.Setenv(k, v)
+		}
+	}
+
+	assert.Equal(t, 7<<20, decoded(t).grpc.ServerMaxRecvMsgSize)
+}

libs/capability/internal/docgen/main.go

@@ -17,7 +17,6 @@ func main() {
 		Roots: []any{
 			&capability.HTTPConfig{},
 			&capability.CapabilitiesConfig{},
-			&capability.GRPCConfig{},
 			&capability.SettingsConfig{},
 		},
 		Tool:        "github.com/smartcontractkit/capabilities/libs/capability/internal/docgen",

libs/capability/registry.go

@@ -17,15 +17,17 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry/remote"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/loop"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 
 	p2ptypes "github.com/smartcontractkit/libocr/ragep2p/types"
 )
 
-func newRegistry(lggr logger.Logger, cfg CapabilitiesConfig, grpcCfg GRPCConfig) (*registryService, error) {
+func newRegistry(lggr logger.Logger, cfg CapabilitiesConfig, grpcCfg GRPCConfig, grpcOpts loop.GRPCOpts) (*registryService, error) {
 	r := &registryService{local: registry.NewRegistry(lggr)}
 
 	if cfg.ProxyURL != "" {
+		dial := dialer(grpcOpts)
 		conn, err := dial(cfg.ProxyURL)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create registry proxy client for %s: %w", cfg.ProxyURL, err)
@@ -36,7 +38,8 @@ func newRegistry(lggr logger.Logger, cfg CapabilitiesConfig, grpcCfg GRPCConfig)
 			Listen: func() (net.Listener, error) {
 				return net.Listen("tcp", net.JoinHostPort(grpcCfg.AdvertiseHost, strconv.Itoa(int(0))))
 			},
-			Connect: dial,
+			Connect:       dial,
+			ServerFactory: func() *grpc.Server { return grpcOpts.NewServer(nil) },
 		})
 		if err != nil {
 			_ = conn.Close()
@@ -51,13 +54,17 @@ func newRegistry(lggr logger.Logger, cfg CapabilitiesConfig, grpcCfg GRPCConfig)
 	return r, nil
 }
 
-func dial(target string) (*grpc.ClientConn, error) {
-	return grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials()))
+// dialer makes gRPC clients with the interceptors in opts. opts has no transport credentials, so
+// dialer adds insecure credentials.
+func dialer(opts loop.GRPCOpts) func(target string) (*grpc.ClientConn, error) {
+	return func(target string) (*grpc.ClientConn, error) {
+		return grpc.NewClient(target, append(slices.Clone(opts.DialOpts), grpc.WithTransportCredentials(insecure.NewCredentials()))...)
+	}
 }
 
 // startRegistry builds the registry and starts it.
-func startRegistry(ctx context.Context, lggr logger.Logger, cfg CapabilitiesConfig, grpcCfg GRPCConfig) (*registryService, error) {
-	r, err := newRegistry(lggr, cfg, grpcCfg)
+func startRegistry(ctx context.Context, lggr logger.Logger, cfg CapabilitiesConfig, grpcCfg GRPCConfig, grpcOpts loop.GRPCOpts) (*registryService, error) {
+	r, err := newRegistry(lggr, cfg, grpcCfg, grpcOpts)
 	if err != nil {
 		return nil, err
 	}

libs/capability/registry_test.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
@@ -14,17 +15,24 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry/remote"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/loop"
 )
 
 var testGRPC = GRPCConfig{AdvertiseHost: defaultHost}
 
+// newTestGRPCOpts makes gRPC options that register their metrics on reg. Each test must use a new
+// reg. Then the test can read its own metrics.
+func newTestGRPCOpts(t *testing.T, reg *prometheus.Registry) loop.GRPCOpts {
+	return loop.GRPCOptsConfig{Registerer: reg}.New(logger.Test(t))
+}
+
 // TestNewRegistryDoesNotDial is the property the boot sequence rests on: the node starts this
 // process and the two race, so a registry proxy that is not up yet has to delay the first lookup
 // rather than fail the run.
 //
 // Nothing is listening on the port below, and building the registry still succeeds.
 func TestNewRegistryDoesNotDial(t *testing.T) {
-	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: "localhost:1", CapabilityDonID: 1}, testGRPC)
+	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: "localhost:1", CapabilityDonID: 1}, testGRPC, newTestGRPCOpts(t, prometheus.NewRegistry()))
 
 	require.NoError(t, err, "an unreachable proxy should not fail construction")
 	require.NotNil(t, reg.fallback)
@@ -42,7 +50,7 @@ func TestNewRegistryDoesNotDial(t *testing.T) {
 func TestNewRegistryDoesNotValidateTheTarget(t *testing.T) {
 	for _, target := range []string{"!://not a target", "unknownscheme:///x"} {
 		t.Run(target, func(t *testing.T) {
-			reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: target}, testGRPC)
+			reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: target}, testGRPC, newTestGRPCOpts(t, prometheus.NewRegistry()))
 			require.NoError(t, err)
 			require.NoError(t, reg.Start(t.Context()))
 			assert.NoError(t, reg.Close())
@@ -56,7 +64,7 @@ func TestNewRegistryDoesNotValidateTheTarget(t *testing.T) {
 // The metadata calls failing is the point rather than a shortcoming. A process holding capability
 // values has no way to know which DONs exist, so saying so beats answering with something invented.
 func TestNewRegistryWithoutAProxyIsLocalOnly(t *testing.T) {
-	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{CapabilityDonID: 1}, testGRPC)
+	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{CapabilityDonID: 1}, testGRPC, newTestGRPCOpts(t, prometheus.NewRegistry()))
 	require.NoError(t, err)
 	t.Cleanup(func() { assert.NoError(t, reg.Close()) })
 	require.NoError(t, reg.Start(t.Context()))
@@ -71,7 +79,7 @@ func TestNewRegistryWithoutAProxyIsLocalOnly(t *testing.T) {
 // TestRegistryAddHoldsUntilClose covers the local half: this process's registry resolves what was
 // added to it, and stops doing so once it closes.
 func TestRegistryAddHoldsUntilClose(t *testing.T) {
-	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{}, testGRPC)
+	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{}, testGRPC, newTestGRPCOpts(t, prometheus.NewRegistry()))
 	require.NoError(t, err)
 	require.NoError(t, reg.Start(t.Context()))
 
@@ -98,7 +106,7 @@ func TestRegistryAddHoldsUntilClose(t *testing.T) {
 func TestRegistryAddAnnouncesToTheNode(t *testing.T) {
 	node := newSpyRegistry(t)
 
-	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: serveNodeRegistry(t, node)}, testGRPC)
+	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: serveNodeRegistry(t, node)}, testGRPC, newTestGRPCOpts(t, prometheus.NewRegistry()))
 	require.NoError(t, err)
 	require.NoError(t, reg.Start(t.Context()))
 
@@ -115,6 +123,57 @@ func TestRegistryAddAnnouncesToTheNode(t *testing.T) {
 	assert.Equal(t, []string{fakeID}, node.removedIDs(), "shutdown should take the announcement back")
 }
 
+// TestRegistryDialsWithTheRunsInterceptors makes sure that calls to the node use the gRPC options.
+// The client metrics must count the call.
+func TestRegistryDialsWithTheRunsInterceptors(t *testing.T) {
+	metrics := prometheus.NewRegistry()
+	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: serveNodeRegistry(t, newSpyRegistry(t))}, testGRPC, newTestGRPCOpts(t, metrics))
+	require.NoError(t, err)
+	require.NoError(t, reg.Start(t.Context()))
+	t.Cleanup(func() { assert.NoError(t, reg.Close()) })
+
+	require.NoError(t, reg.Add(t.Context(), newFake()))
+
+	assert.Positive(t, counterTotal(t, metrics, "grpc_client_handled_total"), "the announcement should be counted")
+}
+
+// TestRegistryServesWithTheRunsInterceptors makes sure that calls from the node use the gRPC options.
+// The server metrics must count the call.
+func TestRegistryServesWithTheRunsInterceptors(t *testing.T) {
+	node := newSpyRegistry(t)
+	metrics := prometheus.NewRegistry()
+	reg, err := newRegistry(logger.Test(t), CapabilitiesConfig{ProxyURL: serveNodeRegistry(t, node)}, testGRPC, newTestGRPCOpts(t, metrics))
+	require.NoError(t, err)
+	require.NoError(t, reg.Start(t.Context()))
+	t.Cleanup(func() { assert.NoError(t, reg.Close()) })
+	require.NoError(t, reg.Add(t.Context(), newFake()))
+
+	held, err := node.Get(t.Context(), fakeID)
+	require.NoError(t, err)
+	_, err = held.Info(t.Context())
+	require.NoError(t, err)
+
+	assert.Positive(t, counterTotal(t, metrics, "grpc_server_handled_total"), "the node's call should be counted")
+}
+
+// counterTotal adds the values of all series of the counter with this name.
+func counterTotal(t *testing.T, g prometheus.Gatherer, name string) float64 {
+	t.Helper()
+	families, err := g.Gather()
+	require.NoError(t, err)
+
+	var total float64
+	for _, f := range families {
+		if f.GetName() != name {
+			continue
+		}
+		for _, m := range f.GetMetric() {
+			total += m.GetCounter().GetValue()
+		}
+	}
+	return total
+}
+
 // spyRegistry is the node's registry, recording what is announced to it and what is taken back.
 type spyRegistry struct {
 	registry.CapabilitiesRegistry
@@ -167,7 +226,7 @@ func serveNodeRegistry(t *testing.T, reg registry.CapabilitiesRegistry) string {
 	transport, err := remote.NewListenerTransport(remote.ListenerTransportParams{
 		Lggr:    lggr,
 		Listen:  func() (net.Listener, error) { return net.Listen("tcp", "localhost:0") },
-		Connect: dial,
+		Connect: dialer(loop.GRPCOpts{}),
 	})
 	require.NoError(t, err)
 

libs/capability/run.go

@@ -11,6 +11,7 @@ import (
 	"strconv"
 	"syscall"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/samber/do/v2"
 	"github.com/spf13/cobra"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
@@ -101,6 +102,12 @@ func RunErr[T Capability](ctx context.Context, lggr logger.Logger, ctor func(do.
 
 type Option func(*config)
 
+// withRegisterer gives a run its own metrics registry. Tests start many runs in one process.
+// The default registry panics if you register the same metrics two times.
+func withRegisterer(r prometheus.Registerer) Option {
+	return func(c *config) { c.grpc.Registerer = r }
+}
+
 // WithOtelViews sets otel metric views - histogram bucket boundaries, typically - on the beholder
 // client this process reports through.
 //

libs/capability/run_test.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/jonboulle/clockwork"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/samber/do/v2"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry"
 	"github.com/stretchr/testify/assert"
@@ -96,7 +97,7 @@ func execute(t *testing.T, lggr logger.Logger, ctor func(do.Injector) (Capabilit
 	defer cancel()
 
 	done := make(chan error, 1)
-	go func() { done <- RunErr(ctx, lggr, ctor) }()
+	go func() { done <- RunErr(ctx, lggr, ctor, withRegisterer(prometheus.NewRegistry())) }()
 
 	if port == 0 {
 		// Nothing that serves: a rejected constructor, or a bare invocation that prints help.

libs/capability/runner.go

@@ -41,13 +41,17 @@ func run(ctx context.Context, lggr logger.Logger, name string, cfg *config, newC
 	}
 	svcs = append(svcs, telemetry)
 
-	settings, err := startSettings(ctx, lggr, cfg.settings)
+	// Start telemetry before you make the gRPC options. The gRPC tracing and metrics read the
+	// telemetry providers when you make them.
+	grpcOpts := cfg.grpc.New(lggr)
+
+	settings, err := startSettings(ctx, lggr, cfg.settings, grpcOpts)
 	if err != nil {
 		return fmt.Errorf("failed to start settings: %w", err)
 	}
 	svcs = append(svcs, settings)
 
-	_, started, err := StartCapabilities(ctx, lggr, base, total, newCapability, withNode(capabilities, cfg.grpc, settings.values))
+	_, started, err := StartCapabilities(ctx, lggr, base, total, newCapability, withNode(capabilities, cfg.grpc, &grpcOpts, settings.values))
 	svcs = append(svcs, started...)
 	if err != nil {
 		return err

libs/capability/runner_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/samber/do/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -24,6 +25,7 @@ func runArgs(t *testing.T) (*config, uint16) {
 
 	port := uint16(freePort(t))
 	cfg := defaultConfig()
+	cfg.grpc.Registerer = prometheus.NewRegistry()
 	cfg.observability.http.Port = port
 	// A stub the registrar's announcement can land on: registering the capability dials the
 	// proxy, so it has to be a real listener rather than merely a lazy one.