Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Per-service fx-ified OTEL tracing #2896

Merged
merged 27 commits into from
Jun 22, 2022
Merged

Per-service fx-ified OTEL tracing #2896

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
2e73608
WIP per-service fx-ified OTEL tracing
May 24, 2022
30a61ad
Ignore .envrc for those using direnv
May 27, 2022
4597795
Move OTEL fx code into temporal/fx.go
May 27, 2022
abce48d
Put otelgrpc interceptor on frontend service
May 27, 2022
d2f34ee
The INSECURE property can be inferred from the URI scheme
May 27, 2022
ff390ff
Find service name through DI rather than parameter
May 31, 2022
42cc3f5
Use the resource info detectors OTEL provides
May 31, 2022
25e1978
Build otel exporter configuration from yaml
Jun 3, 2022
b1fedf9
rebase fixups
Jun 3, 2022
06e63ee
Lift some literals up into constants
Jun 3, 2022
bac0b2f
License comments
Jun 3, 2022
aabe9a8
Comments!
Jun 4, 2022
341112a
License header
Jun 4, 2022
1d87471
Unit tests
Jun 4, 2022
f858b34
Comments and licenses for linters
Jun 4, 2022
7148c78
Wrap exporters referencing shared grpc.ClientConns
Jun 5, 2022
7784a8d
Get onebox fx working again
Jun 5, 2022
7a86512
Mark span exporters as an optional dependency
Jun 5, 2022
db89208
Put type declarations where they belong
Jun 6, 2022
4cd4aed
Shrink PR
Jun 6, 2022
255136d
Plumb through the client interceptors
Jun 6, 2022
e685282
Lift more constants out from inline
Jun 6, 2022
8d8f54b
Rename 'proto' -> 'protocol' in OTEL config
Jun 7, 2022
e823655
Disable tracing by default
Jun 7, 2022
c2f26d0
Rename sdktrace -> otelsdktrace
Jun 7, 2022
fde3ea7
Coalesce go.mod require sections
Jun 22, 2022
2803188
Prefer otelsdktrace to sdktrace
Jun 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,3 +28,5 @@
# Fossa
.fossa.yml

# direnv
.envrc
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just fyi, my strategy is to put an .envrc in the directory above this repo, so I can share it with the other temporal repos too

3 changes: 3 additions & 0 deletions common/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ import (
"go.temporal.io/server/common/masker"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
"go.temporal.io/server/common/telemetry"
)

type (
Expand All @@ -64,6 +65,8 @@ type (
DynamicConfigClient *dynamicconfig.FileBasedClientConfig `yaml:"dynamicConfigClient"`
// NamespaceDefaults is the default config for every namespace
NamespaceDefaults NamespaceDefaults `yaml:"namespaceDefaults"`
// ExporterConfig allows the specification of process-wide OTEL exporters
ExporterConfig telemetry.ExportConfig `yaml:"otel"`
}

// Service contains the service specific config items
Expand Down
15 changes: 14 additions & 1 deletion common/resource/fx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import (
"time"

"go.uber.org/fx"
"google.golang.org/grpc"

"go.temporal.io/api/workflowservice/v1"

Expand Down Expand Up @@ -62,6 +63,7 @@ import (
"go.temporal.io/server/common/rpc/encryption"
"go.temporal.io/server/common/sdk"
"go.temporal.io/server/common/searchattribute"
"go.temporal.io/server/common/telemetry"
)

type (
Expand Down Expand Up @@ -416,7 +418,18 @@ func RPCFactoryProvider(
tlsConfigProvider encryption.TLSConfigProvider,
dc *dynamicconfig.Collection,
clusterMetadata *cluster.Config,
traceInterceptor telemetry.ClientTraceInterceptor,
) common.RPCFactory {
svcCfg := cfg.Services[string(svcName)]
return rpc.NewFactory(&svcCfg.RPC, string(svcName), logger, tlsConfigProvider, dc, clusterMetadata)
return rpc.NewFactory(
&svcCfg.RPC,
string(svcName),
logger,
tlsConfigProvider,
dc,
clusterMetadata,
[]grpc.UnaryClientInterceptor{
grpc.UnaryClientInterceptor(traceInterceptor),
},
)
}
11 changes: 7 additions & 4 deletions common/rpc/grpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ const (
// The hostName syntax is defined in
// https://github.com/grpc/grpc/blob/master/doc/naming.md.
// e.g. to use dns resolver, a "dns:///" prefix should be applied to the target.
func Dial(hostName string, tlsConfig *tls.Config, logger log.Logger) (*grpc.ClientConn, error) {
func Dial(hostName string, tlsConfig *tls.Config, logger log.Logger, interceptors ...grpc.UnaryClientInterceptor) (*grpc.ClientConn, error) {
// Default to insecure
grpcSecureOpt := grpc.WithInsecure()
if tlsConfig != nil {
Expand All @@ -83,9 +83,12 @@ func Dial(hostName string, tlsConfig *tls.Config, logger log.Logger) (*grpc.Clie
grpcSecureOpt,
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxInternodeRecvPayloadSize)),
grpc.WithChainUnaryInterceptor(
versionHeadersInterceptor,
metrics.NewClientMetricsTrailerPropagatorInterceptor(logger),
errorInterceptor,
append(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the order important here?
If so we should probably leave a comment as such.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's sort of a bigger problem in that we have some interceptors that are hard-coded to exist and another set that comes from runtime parameters so if there was some sort of important ordering dependency then it would be difficult to resolve.

That said, in this case where I happen to know that the interceptor slice contains just the trace interceptor (or is empty) then the order isn't all that imporant.

interceptors,
versionHeadersInterceptor,
metrics.NewClientMetricsTrailerPropagatorInterceptor(logger),
errorInterceptor,
)...,
),
grpc.WithDefaultServiceConfig(DefaultServiceConfig),
grpc.WithDisableServiceConfig(),
Expand Down
21 changes: 12 additions & 9 deletions common/rpc/rpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,8 +50,9 @@ type RPCFactory struct {
clusterMetadata *cluster.Config

sync.Mutex
grpcListener net.Listener
tlsFactory encryption.TLSConfigProvider
grpcListener net.Listener
tlsFactory encryption.TLSConfigProvider
clientInterceptors []grpc.UnaryClientInterceptor
}

// NewFactory builds a new RPCFactory
Expand All @@ -63,14 +64,16 @@ func NewFactory(
tlsProvider encryption.TLSConfigProvider,
dc *dynamicconfig.Collection,
clusterMetadata *cluster.Config,
clientInterceptors []grpc.UnaryClientInterceptor,
) *RPCFactory {
return &RPCFactory{
config: cfg,
serviceName: sName,
logger: logger,
dc: dc,
tlsFactory: tlsProvider,
clusterMetadata: clusterMetadata,
config: cfg,
serviceName: sName,
logger: logger,
dc: dc,
tlsFactory: tlsProvider,
clusterMetadata: clusterMetadata,
clientInterceptors: clientInterceptors,
}
}

Expand Down Expand Up @@ -225,7 +228,7 @@ func (d *RPCFactory) CreateInternodeGRPCConnection(hostName string) *grpc.Client
}

func (d *RPCFactory) dial(hostName string, tlsClientConfig *tls.Config) *grpc.ClientConn {
connection, err := Dial(hostName, tlsClientConfig, d.logger)
connection, err := Dial(hostName, tlsClientConfig, d.logger, d.clientInterceptors...)
if err != nil {
d.logger.Fatal("Failed to create gRPC connection", tag.Error(err))
return nil
Expand Down
27 changes: 15 additions & 12 deletions common/rpc/test/rpc_localstore_tls_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ import (
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"

"go.temporal.io/server/common/config"
Expand All @@ -50,6 +51,8 @@ const (
frontendServerCertSerialNumber = 150
)

var noExtraInterceptors = []grpc.UnaryClientInterceptor{}

type localStoreRPCSuite struct {
*require.Assertions
suite.Suite
Expand Down Expand Up @@ -128,7 +131,7 @@ func (s *localStoreRPCSuite) SetupSuite() {

provider, err := encryption.NewTLSConfigProviderFromConfig(serverCfgInsecure.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
insecureFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
insecureFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(insecureFactory)
s.insecureRPCFactory = i(insecureFactory)

Expand Down Expand Up @@ -336,22 +339,22 @@ func (s *localStoreRPCSuite) setupFrontend() {

provider, err := encryption.NewTLSConfigProviderFromConfig(localStoreMutualTLS.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
frontendMutualTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
frontendMutualTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(frontendMutualTLSFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreServerTLS.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
frontendServerTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
frontendServerTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(frontendServerTLSFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreMutualTLSSystemWorker.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
frontendSystemWorkerMutualTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
frontendSystemWorkerMutualTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(frontendSystemWorkerMutualTLSFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreMutualTLSWithRefresh.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
frontendMutualTLSRefreshFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
frontendMutualTLSRefreshFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(frontendMutualTLSRefreshFactory)

s.frontendMutualTLSRPCFactory = f(frontendMutualTLSFactory)
Expand All @@ -365,21 +368,21 @@ func (s *localStoreRPCSuite) setupFrontend() {
s.frontendRollingCerts,
s.dynamicCACertPool,
s.wrongCACertPool)
dynamicServerTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, s.dynamicConfigProvider, dynamicconfig.NewNoopCollection(), clusterMetadata)
dynamicServerTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, s.dynamicConfigProvider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.frontendDynamicTLSFactory = f(dynamicServerTLSFactory)
s.internodeDynamicTLSFactory = i(dynamicServerTLSFactory)

s.frontendMutualTLSRPCRefreshFactory = f(frontendMutualTLSRefreshFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreRootCAForceTLS.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
frontendRootCAForceTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
frontendRootCAForceTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(frontendServerTLSFactory)
s.frontendConfigRootCAForceTLSFactory = f(frontendRootCAForceTLSFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreMutualTLSRemoteCluster.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
remoteClusterMutualTLSRPCFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
remoteClusterMutualTLSRPCFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(remoteClusterMutualTLSRPCFactory)
s.remoteClusterMutualTLSRPCFactory = r(remoteClusterMutualTLSRPCFactory)
}
Expand Down Expand Up @@ -415,22 +418,22 @@ func (s *localStoreRPCSuite) setupInternode() {

provider, err := encryption.NewTLSConfigProviderFromConfig(localStoreMutualTLS.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
internodeMutualTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
internodeMutualTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(internodeMutualTLSFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreServerTLS.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
internodeServerTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
internodeServerTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(internodeServerTLSFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreAltMutualTLS.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
internodeMutualAltTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
internodeMutualAltTLSFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(internodeMutualAltTLSFactory)

provider, err = encryption.NewTLSConfigProviderFromConfig(localStoreMutualTLSWithRefresh.TLS, metrics.NoopClient, s.logger, nil)
s.NoError(err)
internodeMutualTLSRefreshFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata)
internodeMutualTLSRefreshFactory := rpc.NewFactory(rpcTestCfgDefault, "tester", s.logger, provider, dynamicconfig.NewNoopCollection(), clusterMetadata, noExtraInterceptors)
s.NotNil(internodeMutualTLSRefreshFactory)

s.internodeMutualTLSRPCFactory = i(internodeMutualTLSFactory)
Expand Down
Loading