[googlecloudpubsubreceiver] add configs to specify credentials info

receiver/googlecloudpubsubreceiver/README.md

@@ -30,6 +30,8 @@ The following configuration options are supported:
   or switching between [global and regional service endpoints](https://cloud.google.com/pubsub/docs/reference/service_apis_overview#service_endpoints).
 * `insecure` (Optional): allows performing “insecure” SSL connections and transfers, useful when connecting to a local
    emulator instance. Only has effect if Endpoint is not ""
+* `credentials_file_path` (Optional): Path to a credentials file to use to connect.
+* `credentials_json` (Optional): Credentials JSON to use to connect.
 
 ```yaml
 receivers:

receiver/googlecloudpubsubreceiver/config.go

@@ -4,6 +4,7 @@
 package googlecloudpubsubreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudpubsubreceiver"
 
 import (
+	"encoding/json"
 	"fmt"
 	"regexp"
 
@@ -13,6 +14,10 @@ import (
 var subscriptionMatcher = regexp.MustCompile(`projects/[a-z][a-z0-9\-]*/subscriptions/`)
 
 type Config struct {
+	// Path to a credentials file to connect to the service
+	CredentialsFilePath string `mapstructure:"credentials_file_path"`
+	// JSON string containing credentials data to connect to the service
+	CredentialsJSON string `mapstructure:"credentials_json"`
 	// Google Cloud Project ID where the Pubsub client will connect to
 	ProjectID string `mapstructure:"project"`
 	// User agent that will be used by the Pubsub client to connect to the service
@@ -90,5 +95,16 @@ func (config *Config) validate() error {
 	default:
 		return fmt.Errorf("compression %v is not supported.  supported compression formats include [gzip]", config.Compression)
 	}
+
+	if config.CredentialsJSON != "" && config.CredentialsFilePath != "" {
+		return fmt.Errorf("only one of credentials_file_path and credentials_json must be specified")
+	}
+
+	if config.CredentialsJSON != "" {
+		if !json.Valid([]byte(config.CredentialsJSON)) {
+			return fmt.Errorf("credentials_json is an invalid JSON")
+		}
+	}
+
 	return nil
 }

receiver/googlecloudpubsubreceiver/config_test.go

@@ -130,3 +130,39 @@ func TestLogConfigValidation(t *testing.T) {
 	c.Encoding = "otlp_proto_log"
 	assert.NoError(t, c.validateForLog())
 }
+
+func TestConfigCredentialsValidation(t *testing.T) {
+	factory := NewFactory()
+	c := factory.CreateDefaultConfig().(*Config)
+	c.Subscription = "projects/my-project/subscriptions/my-subscription"
+
+	dummyCredentialsJSON := `{
+		"type": "service_account",
+		"project_id": "my-project",
+		"private_key_id": "d41d8cd98f00b204e9800998ecf8427e",
+		"private_key": "-----BEGIN PRIVATE KEY-----\njciFXV3SnwCNXBLSykSNBQ\n-----END PRIVATE KEY-----\n",
+		"client_email": "[email protected]",
+		"client_id": "87689569482041111",
+		"auth_uri": "https://accounts.google.com/o/oauth2/auth",
+		"token_uri": "https://oauth2.googleapis.com/token",
+		"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+		"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/my-project.iam.gserviceaccount.com",
+		"universe_domain": "googleapis.com"
+	  }`
+
+	c.CredentialsFilePath = "/home/service-principal-01ba5c7dde9b.json"
+	c.CredentialsJSON = ""
+	assert.NoError(t, c.validate())
+
+	c.CredentialsFilePath = ""
+	c.CredentialsJSON = dummyCredentialsJSON
+	assert.NoError(t, c.validate())
+
+	c.CredentialsFilePath = "/home/service-principal-01ba5c7dde9b.json"
+	c.CredentialsJSON = dummyCredentialsJSON
+	assert.Error(t, c.validate(), "only one of")
+
+	c.CredentialsFilePath = ""
+	c.CredentialsJSON = "invalid JSON"
+	assert.Error(t, c.validate(), "credentials_json is an invalid JSON")
+}

receiver/googlecloudpubsubreceiver/receiver.go

@@ -84,6 +84,13 @@ func (receiver *pubsubReceiver) generateClientOptions() (copts []option.ClientOp
 			copts = append(copts, option.WithEndpoint(receiver.config.Endpoint))
 		}
 	}
+	if receiver.config.CredentialsFilePath != "" {
+		copts = append(copts, option.WithCredentialsFile(receiver.config.CredentialsFilePath))
+	}
+	if receiver.config.CredentialsJSON != "" {
+		copts = append(copts, option.WithCredentialsJSON([]byte(receiver.config.CredentialsJSON)))
+	}
+
 	return copts
 }
 
@@ -100,6 +107,7 @@ func (receiver *pubsubReceiver) Start(ctx context.Context, _ component.Host) err
 			startErr = fmt.Errorf("failed creating the gRPC client to Pubsub: %w", err)
 			return
 		}
+
 		receiver.client = client
 
 		err = receiver.createReceiverHandler(ctx)