O11Y: Added support for custom tags (#5565)

* O11Y: Added support for custom tags

Author: zasweq

gcp/observability/config.go

@@ -28,8 +28,6 @@ import (
 
 	gcplogging "cloud.google.com/go/logging"
 	"golang.org/x/oauth2/google"
-	configpb "google.golang.org/grpc/gcp/observability/internal/config"
-	"google.golang.org/protobuf/encoding/protojson"
 )
 
 const (
@@ -41,6 +39,69 @@ const (
 
 var logFilterPatternRegexp = regexp.MustCompile(logFilterPatternRegexpStr)
 
+// logFilter represents a method logging configuration.
+type logFilter struct {
+	// Pattern is a string which can select a group of method names. By
+	// default, the Pattern is an empty string, matching no methods.
+	//
+	// Only "*" Wildcard is accepted for Pattern. A Pattern is in the form
+	// of <service>/<method> or just a character "*" .
+	//
+	// If the Pattern is "*", it specifies the defaults for all the
+	// services; If the Pattern is <service>/*, it specifies the defaults
+	// for all methods in the specified service <service>; If the Pattern is
+	// */<method>, this is not supported.
+	//
+	// Examples:
+	//  - "Foo/Bar" selects only the method "Bar" from service "Foo"
+	//  - "Foo/*" selects all methods from service "Foo"
+	//  - "*" selects all methods from all services.
+	Pattern string `json:"pattern,omitempty"`
+	// HeaderBytes is the number of bytes of each header to log. If the size of
+	// the header is greater than the defined limit, content past the limit will
+	// be truncated. The default value is 0.
+	HeaderBytes int32 `json:"header_bytes,omitempty"`
+	// MessageBytes is the number of bytes of each message to log. If the size
+	// of the message is greater than the defined limit, content pass the limit
+	// will be truncated. The default value is 0.
+	MessageBytes int32 `json:"message_bytes,omitempty"`
+}
+
+// config is configuration for observability behaviors. By default, no
+// configuration is required for tracing/metrics/logging to function. This
+// config captures the most common knobs for gRPC users. It's always possible to
+// override with explicit config in code.
+type config struct {
+	// EnableCloudTrace represents whether the tracing data upload to
+	// CloudTrace should be enabled or not.
+	EnableCloudTrace bool `json:"enable_cloud_trace,omitempty"`
+	// EnableCloudMonitoring represents whether the metrics data upload to
+	// CloudMonitoring should be enabled or not.
+	EnableCloudMonitoring bool `json:"enable_cloud_monitoring,omitempty"`
+	// EnableCloudLogging represents Whether the logging data upload to
+	// CloudLogging should be enabled or not.
+	EnableCloudLogging bool `json:"enable_cloud_logging,omitempty"`
+	// DestinationProjectID is the destination GCP project identifier for the
+	// uploading log entries. If empty, the gRPC Observability plugin will
+	// attempt to fetch the project_id from the GCP environment variables, or
+	// from the default credentials.
+	DestinationProjectID string `json:"destination_project_id,omitempty"`
+	// LogFilters is a list of method config. The order matters here - the first
+	// Pattern which matches the current method will apply the associated config
+	// options in the logFilter. Any other logFilter that also matches that
+	// comes later will be ignored. So a logFilter of "*/*" should appear last
+	// in this list.
+	LogFilters []logFilter `json:"log_filters,omitempty"`
+	// GlobalTraceSamplingRate is the global setting that controls the
+	// probability of a RPC being traced. For example, 0.05 means there is a 5%
+	// chance for a RPC to be traced, 1.0 means trace every call, 0 means don’t
+	// start new traces.
+	GlobalTraceSamplingRate float64 `json:"global_trace_sampling_rate,omitempty"`
+	// CustomTags a list of custom tags that will be attached to every log
+	// entry.
+	CustomTags map[string]string `json:"custom_tags,omitempty"`
+}
+
 // fetchDefaultProjectID fetches the default GCP project id from environment.
 func fetchDefaultProjectID(ctx context.Context) string {
 	// Step 1: Check ENV var
@@ -62,25 +123,25 @@ func fetchDefaultProjectID(ctx context.Context) string {
 	return credentials.ProjectID
 }
 
-func validateFilters(config *configpb.ObservabilityConfig) error {
-	for _, filter := range config.GetLogFilters() {
+func validateFilters(config *config) error {
+	for _, filter := range config.LogFilters {
 		if filter.Pattern == "*" {
 			continue
 		}
 		match := logFilterPatternRegexp.FindStringSubmatch(filter.Pattern)
 		if match == nil {
-			return fmt.Errorf("invalid log filter pattern: %v", filter.Pattern)
+			return fmt.Errorf("invalid log filter Pattern: %v", filter.Pattern)
 		}
 	}
 	return nil
 }
 
 // unmarshalAndVerifyConfig unmarshals a json string representing an
-// observability config into its protobuf format, and also verifies the
+// observability config into its internal go format, and also verifies the
 // configuration's fields for validity.
-func unmarshalAndVerifyConfig(rawJSON json.RawMessage) (*configpb.ObservabilityConfig, error) {
-	var config configpb.ObservabilityConfig
-	if err := protojson.Unmarshal(rawJSON, &config); err != nil {
+func unmarshalAndVerifyConfig(rawJSON json.RawMessage) (*config, error) {
+	var config config
+	if err := json.Unmarshal(rawJSON, &config); err != nil {
 		return nil, fmt.Errorf("error parsing observability config: %v", err)
 	}
 	if err := validateFilters(&config); err != nil {
@@ -93,7 +154,7 @@ func unmarshalAndVerifyConfig(rawJSON json.RawMessage) (*configpb.ObservabilityC
 	return &config, nil
 }
 
-func parseObservabilityConfig() (*configpb.ObservabilityConfig, error) {
+func parseObservabilityConfig() (*config, error) {
 	if fileSystemPath := os.Getenv(envObservabilityConfigJSON); fileSystemPath != "" {
 		content, err := ioutil.ReadFile(fileSystemPath) // TODO: Switch to os.ReadFile once dropped support for go 1.15
 		if err != nil {
@@ -107,14 +168,14 @@ func parseObservabilityConfig() (*configpb.ObservabilityConfig, error) {
 	return nil, nil
 }
 
-func ensureProjectIDInObservabilityConfig(ctx context.Context, config *configpb.ObservabilityConfig) error {
-	if config.GetDestinationProjectId() == "" {
+func ensureProjectIDInObservabilityConfig(ctx context.Context, config *config) error {
+	if config.DestinationProjectID == "" {
 		// Try to fetch the GCP project id
 		projectID := fetchDefaultProjectID(ctx)
 		if projectID == "" {
 			return fmt.Errorf("empty destination project ID")
 		}
-		config.DestinationProjectId = projectID
+		config.DestinationProjectID = projectID
 	}
 	return nil
 }

gcp/observability/exporting.go

@@ -22,7 +22,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"os"
 
 	gcplogging "cloud.google.com/go/logging"
 	grpclogrecordpb "google.golang.org/grpc/gcp/observability/internal/logging"
@@ -45,20 +44,19 @@ type cloudLoggingExporter struct {
 	logger    *gcplogging.Logger
 }
 
-func newCloudLoggingExporter(ctx context.Context, projectID string) (*cloudLoggingExporter, error) {
-	c, err := gcplogging.NewClient(ctx, fmt.Sprintf("projects/%v", projectID))
+func newCloudLoggingExporter(ctx context.Context, config *config) (*cloudLoggingExporter, error) {
+	c, err := gcplogging.NewClient(ctx, fmt.Sprintf("projects/%v", config.DestinationProjectID))
 	if err != nil {
 		return nil, fmt.Errorf("failed to create cloudLoggingExporter: %v", err)
 	}
 	defer logger.Infof("Successfully created cloudLoggingExporter")
-	customTags := getCustomTags(os.Environ())
-	if len(customTags) != 0 {
-		logger.Infof("Adding custom tags: %+v", customTags)
+	if len(config.CustomTags) != 0 {
+		logger.Infof("Adding custom tags: %+v", config.CustomTags)
 	}
 	return &cloudLoggingExporter{
-		projectID: projectID,
+		projectID: config.DestinationProjectID,
 		client:    c,
-		logger:    c.Logger("microservices.googleapis.com/observability/grpc", gcplogging.CommonLabels(customTags)),
+		logger:    c.Logger("microservices.googleapis.com/observability/grpc", gcplogging.CommonLabels(config.CustomTags)),
 	}, nil
 }
 

gcp/observability/internal/config/config.pb.go

@@ -27,7 +27,6 @@ import (
 
 	"github.com/google/uuid"
 	binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
-	configpb "google.golang.org/grpc/gcp/observability/internal/config"
 	grpclogrecordpb "google.golang.org/grpc/gcp/observability/internal/logging"
 	iblog "google.golang.org/grpc/internal/binarylog"
 )
@@ -248,7 +247,7 @@ func (l *binaryLogger) Close() {
 	}
 }
 
-func validateExistingMethodLoggerConfig(existing *iblog.MethodLoggerConfig, filter *configpb.ObservabilityConfig_LogFilter) bool {
+func validateExistingMethodLoggerConfig(existing *iblog.MethodLoggerConfig, filter logFilter) bool {
 	// In future, we could add more validations. Currently, we only check if the
 	// new filter configs are different than the existing one, if so, we log a
 	// warning.
@@ -258,7 +257,7 @@ func validateExistingMethodLoggerConfig(existing *iblog.MethodLoggerConfig, filt
 	return existing == nil
 }
 
-func createBinaryLoggerConfig(filters []*configpb.ObservabilityConfig_LogFilter) iblog.LoggerConfig {
+func createBinaryLoggerConfig(filters []logFilter) iblog.LoggerConfig {
 	config := iblog.LoggerConfig{
 		Services: make(map[string]*iblog.MethodLoggerConfig),
 		Methods:  make(map[string]*iblog.MethodLoggerConfig),
@@ -296,8 +295,8 @@ func createBinaryLoggerConfig(filters []*configpb.ObservabilityConfig_LogFilter)
 
 // start is the core logic for setting up the custom binary logging logger, and
 // it's also useful for testing.
-func (l *binaryLogger) start(config *configpb.ObservabilityConfig, exporter loggingExporter) error {
-	filters := config.GetLogFilters()
+func (l *binaryLogger) start(config *config, exporter loggingExporter) error {
+	filters := config.LogFilters
 	if len(filters) == 0 || exporter == nil {
 		// Doing nothing is allowed
 		if exporter != nil {
@@ -318,14 +317,14 @@ func (l *binaryLogger) start(config *configpb.ObservabilityConfig, exporter logg
 	return nil
 }
 
-func (l *binaryLogger) Start(ctx context.Context, config *configpb.ObservabilityConfig) error {
-	if config == nil || !config.GetEnableCloudLogging() {
+func (l *binaryLogger) Start(ctx context.Context, config *config) error {
+	if config == nil || !config.EnableCloudLogging {
 		return nil
 	}
-	if config.GetDestinationProjectId() == "" {
+	if config.DestinationProjectID == "" {
 		return fmt.Errorf("failed to enable CloudLogging: empty destination_project_id")
 	}
-	exporter, err := newCloudLoggingExporter(ctx, config.DestinationProjectId)
+	exporter, err := newCloudLoggingExporter(ctx, config)
 	if err != nil {
 		return fmt.Errorf("unable to create CloudLogging exporter: %v", err)
 	}

gcp/observability/internal/config/config.proto

@@ -64,7 +64,7 @@ func Start(ctx context.Context) error {
 	}
 
 	// Enabling tracing and metrics via OpenCensus
-	if err := startOpenCensus(config, nil); err != nil {
+	if err := startOpenCensus(config); err != nil {
 		return fmt.Errorf("failed to instrument OpenCensus: %v", err)
 	}
 

gcp/observability/logging.go

@@ -34,8 +34,8 @@ import (
 	"go.opencensus.io/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
-	configpb "google.golang.org/grpc/gcp/observability/internal/config"
 	grpclogrecordpb "google.golang.org/grpc/gcp/observability/internal/logging"
+	"google.golang.org/grpc/internal"
 	iblog "google.golang.org/grpc/internal/binarylog"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/leakcheck"
@@ -129,7 +129,6 @@ func (fle *fakeLoggingExporter) Close() error {
 type test struct {
 	t   *testing.T
 	fle *fakeLoggingExporter
-	fe  *fakeOpenCensusExporter
 
 	testServer testgrpc.TestServiceServer // nil means none
 	// srv and srvAddr are set once startServer is called.
@@ -197,7 +196,7 @@ func (te *test) clientConn() *grpc.ClientConn {
 	return te.cc
 }
 
-func (te *test) enablePluginWithConfig(config *configpb.ObservabilityConfig) {
+func (te *test) enablePluginWithConfig(config *config) {
 	// Injects the fake exporter for testing purposes
 	te.fle = &fakeLoggingExporter{t: te.t}
 	defaultLogger = newBinaryLogger(nil)
@@ -208,10 +207,10 @@ func (te *test) enablePluginWithConfig(config *configpb.ObservabilityConfig) {
 }
 
 func (te *test) enablePluginWithCaptureAll() {
-	te.enablePluginWithConfig(&configpb.ObservabilityConfig{
+	te.enablePluginWithConfig(&config{
 		EnableCloudLogging:   true,
-		DestinationProjectId: "fake",
-		LogFilters: []*configpb.ObservabilityConfig_LogFilter{
+		DestinationProjectID: "fake",
+		LogFilters: []logFilter{
 			{
 				Pattern:      "*",
 				HeaderBytes:  infinitySizeBytes,
@@ -223,14 +222,13 @@ func (te *test) enablePluginWithCaptureAll() {
 
 func (te *test) enableOpenCensus() {
 	defaultMetricsReportingInterval = time.Millisecond * 100
-	config := &configpb.ObservabilityConfig{
+	config := &config{
 		EnableCloudLogging:      true,
 		EnableCloudTrace:        true,
 		EnableCloudMonitoring:   true,
 		GlobalTraceSamplingRate: 1.0,
 	}
-	te.fe = &fakeOpenCensusExporter{SeenViews: make(map[string]string), t: te.t}
-	startOpenCensus(config, te.fe)
+	startOpenCensus(config)
 }
 
 func checkEventCommon(t *testing.T, seen *grpclogrecordpb.GrpcLogRecord) {
@@ -522,7 +520,7 @@ func (s) TestLoggingForErrorCall(t *testing.T) {
 func (s) TestEmptyConfig(t *testing.T) {
 	te := newTest(t)
 	defer te.tearDown()
-	te.enablePluginWithConfig(&configpb.ObservabilityConfig{})
+	te.enablePluginWithConfig(&config{})
 	te.startServer(&testServer{})
 	tc := testgrpc.NewTestServiceClient(te.clientConn())
 
@@ -554,10 +552,10 @@ func (s) TestOverrideConfig(t *testing.T) {
 	// most specific one. The third filter allows message payload logging, and
 	// others disabling the message payload logging. We should observe this
 	// behavior latter.
-	te.enablePluginWithConfig(&configpb.ObservabilityConfig{
+	te.enablePluginWithConfig(&config{
 		EnableCloudLogging:   true,
-		DestinationProjectId: "fake",
-		LogFilters: []*configpb.ObservabilityConfig_LogFilter{
+		DestinationProjectID: "fake",
+		LogFilters: []logFilter{
 			{
 				Pattern:      "wont/match",
 				MessageBytes: 0,
@@ -621,10 +619,10 @@ func (s) TestNoMatch(t *testing.T) {
 	// Setting 3 filters, expected to use the second filter. The second filter
 	// allows message payload logging, and others disabling the message payload
 	// logging. We should observe this behavior latter.
-	te.enablePluginWithConfig(&configpb.ObservabilityConfig{
+	te.enablePluginWithConfig(&config{
 		EnableCloudLogging:   true,
-		DestinationProjectId: "fake",
-		LogFilters: []*configpb.ObservabilityConfig_LogFilter{
+		DestinationProjectID: "fake",
+		LogFilters: []logFilter{
 			{
 				Pattern:      "wont/match",
 				MessageBytes: 0,
@@ -661,10 +659,10 @@ func (s) TestNoMatch(t *testing.T) {
 }
 
 func (s) TestRefuseStartWithInvalidPatterns(t *testing.T) {
-	config := &configpb.ObservabilityConfig{
+	config := &config{
 		EnableCloudLogging:   true,
-		DestinationProjectId: "fake",
-		LogFilters: []*configpb.ObservabilityConfig_LogFilter{
+		DestinationProjectID: "fake",
+		LogFilters: []logFilter{
 			{
 				Pattern: ":-)",
 			},
@@ -673,7 +671,7 @@ func (s) TestRefuseStartWithInvalidPatterns(t *testing.T) {
 			},
 		},
 	}
-	configJSON, err := protojson.Marshal(config)
+	configJSON, err := json.Marshal(config)
 	if err != nil {
 		t.Fatalf("failed to convert config to JSON: %v", err)
 	}
@@ -689,7 +687,7 @@ func (s) TestRefuseStartWithInvalidPatterns(t *testing.T) {
 // place in the temporary portion of the file system dependent on system. It
 // also sets the environment variable GRPC_CONFIG_OBSERVABILITY_JSON to point to
 // this created config.
-func createTmpConfigInFileSystem(rawJSON string) (*os.File, error) {
+func createTmpConfigInFileSystem(rawJSON string) (func(), error) {
 	configJSONFile, err := ioutil.TempFile(os.TempDir(), "configJSON-")
 	if err != nil {
 		return nil, fmt.Errorf("cannot create file %v: %v", configJSONFile.Name(), err)
@@ -699,19 +697,23 @@ func createTmpConfigInFileSystem(rawJSON string) (*os.File, error) {
 		return nil, fmt.Errorf("cannot write marshalled JSON: %v", err)
 	}
 	os.Setenv(envObservabilityConfigJSON, configJSONFile.Name())
-	return configJSONFile, nil
+	return func() {
+		configJSONFile.Close()
+		os.Setenv(envObservabilityConfigJSON, "")
+	}, nil
 }
 
 // TestJSONEnvVarSet tests a valid observability configuration specified by the
 // GRPC_CONFIG_OBSERVABILITY_JSON environment variable, whose value represents a
 // file path pointing to a JSON encoded config.
 func (s) TestJSONEnvVarSet(t *testing.T) {
 	configJSON := `{
-		"destinationProjectId": "fake",
-		"logFilters":[{"pattern":"*","headerBytes":1073741824,"messageBytes":1073741824}]
+		"destination_project_id": "fake",
+		"log_filters":[{"pattern":"*","header_bytes":1073741824,"message_bytes":1073741824}]
 	}`
-	configJSONFile, err := createTmpConfigInFileSystem(configJSON)
-	defer configJSONFile.Close()
+	cleanup, err := createTmpConfigInFileSystem(configJSON)
+	defer cleanup()
+
 	if err != nil {
 		t.Fatalf("failed to create config in file system: %v", err)
 	}
@@ -730,27 +732,27 @@ func (s) TestJSONEnvVarSet(t *testing.T) {
 // variable is set and valid.
 func (s) TestBothConfigEnvVarsSet(t *testing.T) {
 	configJSON := `{
-		"destinationProjectId":"fake",
-		"logFilters":[{"pattern":":-)"}, {"pattern":"*"}]
+		"destination_project_id":"fake",
+		"log_filters":[{"pattern":":-)"}, {"pattern_string":"*"}]
 	}`
-	configJSONFile, err := createTmpConfigInFileSystem(configJSON)
-	defer configJSONFile.Close()
+	cleanup, err := createTmpConfigInFileSystem(configJSON)
+	defer cleanup()
 	if err != nil {
 		t.Fatalf("failed to create config in file system: %v", err)
 	}
 	// This configuration should be ignored, as precedence 2.
-	validConfig := &configpb.ObservabilityConfig{
+	validConfig := &config{
 		EnableCloudLogging:   true,
-		DestinationProjectId: "fake",
-		LogFilters: []*configpb.ObservabilityConfig_LogFilter{
+		DestinationProjectID: "fake",
+		LogFilters: []logFilter{
 			{
 				Pattern:      "*",
 				HeaderBytes:  infinitySizeBytes,
 				MessageBytes: infinitySizeBytes,
 			},
 		},
 	}
-	validConfigJSON, err := protojson.Marshal(validConfig)
+	validConfigJSON, err := json.Marshal(validConfig)
 	if err != nil {
 		t.Fatalf("failed to convert config to JSON: %v", err)
 	}
@@ -766,6 +768,7 @@ func (s) TestBothConfigEnvVarsSet(t *testing.T) {
 // a file (or valid configuration).
 func (s) TestErrInFileSystemEnvVar(t *testing.T) {
 	os.Setenv(envObservabilityConfigJSON, "/this-file/does-not-exist")
+	defer os.Setenv(envObservabilityConfigJSON, "")
 	if err := Start(context.Background()); err == nil {
 		t.Fatalf("Invalid file system path not triggering error")
 	}
@@ -783,6 +786,16 @@ func (s) TestNoEnvSet(t *testing.T) {
 func (s) TestOpenCensusIntegration(t *testing.T) {
 	te := newTest(t)
 	defer te.tearDown()
+	fe := &fakeOpenCensusExporter{SeenViews: make(map[string]string), t: te.t}
+
+	defer func(ne func(config *config) (tracingMetricsExporter, error)) {
+		newExporter = ne
+	}(newExporter)
+
+	newExporter = func(config *config) (tracingMetricsExporter, error) {
+		return fe, nil
+	}
+
 	te.enableOpenCensus()
 	te.startServer(&testServer{})
 	tc := testgrpc.NewTestServiceClient(te.clientConn())
@@ -807,17 +820,17 @@ func (s) TestOpenCensusIntegration(t *testing.T) {
 	defer cancel()
 	for ctx.Err() == nil {
 		errs = nil
-		te.fe.mu.RLock()
-		if value := te.fe.SeenViews["grpc.io/client/completed_rpcs"]; value != TypeOpenCensusViewCount {
+		fe.mu.RLock()
+		if value := fe.SeenViews["grpc.io/client/completed_rpcs"]; value != TypeOpenCensusViewCount {
 			errs = append(errs, fmt.Errorf("unexpected type for grpc.io/client/completed_rpcs: %s != %s", value, TypeOpenCensusViewCount))
 		}
-		if value := te.fe.SeenViews["grpc.io/server/completed_rpcs"]; value != TypeOpenCensusViewCount {
+		if value := fe.SeenViews["grpc.io/server/completed_rpcs"]; value != TypeOpenCensusViewCount {
 			errs = append(errs, fmt.Errorf("unexpected type for grpc.io/server/completed_rpcs: %s != %s", value, TypeOpenCensusViewCount))
 		}
-		if te.fe.SeenSpans <= 0 {
-			errs = append(errs, fmt.Errorf("unexpected number of seen spans: %v <= 0", te.fe.SeenSpans))
+		if fe.SeenSpans <= 0 {
+			errs = append(errs, fmt.Errorf("unexpected number of seen spans: %v <= 0", fe.SeenSpans))
 		}
-		te.fe.mu.RUnlock()
+		fe.mu.RUnlock()
 		if len(errs) == 0 {
 			break
 		}
@@ -827,3 +840,52 @@ func (s) TestOpenCensusIntegration(t *testing.T) {
 		t.Fatalf("Invalid OpenCensus export data: %v", errs)
 	}
 }
+
+// TestCustomTagsTracingMetrics verifies that the custom tags defined in our
+// observability configuration and set to two hardcoded values are passed to the
+// function to create an exporter.
+func (s) TestCustomTagsTracingMetrics(t *testing.T) {
+	defer func(ne func(config *config) (tracingMetricsExporter, error)) {
+		newExporter = ne
+	}(newExporter)
+	fe := &fakeOpenCensusExporter{SeenViews: make(map[string]string), t: t}
+	newExporter = func(config *config) (tracingMetricsExporter, error) {
+		ct := config.CustomTags
+		if len(ct) < 1 {
+			t.Fatalf("less than 2 custom tags sent in")
+		}
+		if val, ok := ct["customtag1"]; !ok || val != "wow" {
+			t.Fatalf("incorrect custom tag: got %v, want %v", val, "wow")
+		}
+		if val, ok := ct["customtag2"]; !ok || val != "nice" {
+			t.Fatalf("incorrect custom tag: got %v, want %v", val, "nice")
+		}
+		return fe, nil
+	}
+
+	// This configuration present in file system and it's defined custom tags should make it
+	// to the created exporter.
+	configJSON := `{
+		"destination_project_id": "fake",
+		"enable_cloud_trace": true,
+		"enable_cloud_monitoring": true,
+		"global_trace_sampling_rate": 1.0,
+		"custom_tags":{"customtag1":"wow","customtag2":"nice"}
+	}`
+	cleanup, err := createTmpConfigInFileSystem(configJSON)
+	defer cleanup()
+
+	// To clear globally registered tracing and metrics exporters.
+	defer func() {
+		internal.ClearExtraDialOptions()
+		internal.ClearExtraServerOptions()
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	err = Start(ctx)
+	defer End()
+	if err != nil {
+		t.Fatalf("Start() failed with err: %v", err)
+	}
+}

gcp/observability/observability.go

@@ -29,7 +29,6 @@ import (
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/trace"
 	"google.golang.org/grpc"
-	configpb "google.golang.org/grpc/gcp/observability/internal/config"
 	"google.golang.org/grpc/internal"
 )
 
@@ -38,26 +37,59 @@ var (
 	defaultMetricsReportingInterval = time.Second * 30
 )
 
+func tagsToMonitoringLabels(tags map[string]string) *stackdriver.Labels {
+	labels := &stackdriver.Labels{}
+	for k, v := range tags {
+		labels.Set(k, v, "")
+	}
+	return labels
+}
+
+func tagsToTraceAttributes(tags map[string]string) map[string]interface{} {
+	ta := make(map[string]interface{}, len(tags))
+	for k, v := range tags {
+		ta[k] = v
+	}
+	return ta
+}
+
+type tracingMetricsExporter interface {
+	trace.Exporter
+	view.Exporter
+}
+
+// global to stub out in tests
+var newExporter = newStackdriverExporter
+
+func newStackdriverExporter(config *config) (tracingMetricsExporter, error) {
+	// Create the Stackdriver exporter, which is shared between tracing and stats
+	mr := monitoredresource.Autodetect()
+	logger.Infof("Detected MonitoredResource:: %+v", mr)
+	var err error
+	exporter, err := stackdriver.NewExporter(stackdriver.Options{
+		ProjectID:               config.DestinationProjectID,
+		MonitoredResource:       mr,
+		DefaultMonitoringLabels: tagsToMonitoringLabels(config.CustomTags),
+		DefaultTraceAttributes:  tagsToTraceAttributes(config.CustomTags),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Stackdriver exporter: %v", err)
+	}
+	return exporter, nil
+}
+
 // This method accepts config and exporter; the exporter argument is exposed to
 // assist unit testing of the OpenCensus behavior.
-func startOpenCensus(config *configpb.ObservabilityConfig, exporter interface{}) error {
+func startOpenCensus(config *config) error {
 	// If both tracing and metrics are disabled, there's no point inject default
 	// StatsHandler.
 	if config == nil || (!config.EnableCloudTrace && !config.EnableCloudMonitoring) {
 		return nil
 	}
 
-	if exporter == nil {
-		// Create the Stackdriver exporter, which is shared between tracing and stats
-		mr := monitoredresource.Autodetect()
-		logger.Infof("Detected MonitoredResource:: %+v", mr)
-		var err error
-		if exporter, err = stackdriver.NewExporter(stackdriver.Options{
-			ProjectID:         config.DestinationProjectId,
-			MonitoredResource: mr,
-		}); err != nil {
-			return fmt.Errorf("failed to create Stackdriver exporter: %v", err)
-		}
+	exporter, err := newExporter(config)
+	if err != nil {
+		return err
 	}
 
 	var so trace.StartOptions