Enable keyless access to GCP pubsub with workload Identity Federation

[pallabkroy]

Support workload identity federation to authenticate with GCP pubsub: At present, logstash-output-google_pubsub uses service account key to access GCP pubsub which introduces the risks associated with managing long-lived keys for application. We can avoid this risk by using GCP workload identity federation instead of service account key. Can we get this feature for logstash-output-google_pubsub plugin which will use workload identity federation to access GCP pubsub ?

Details of the GCP workload identity federation is provided in that link : https://cloud.google.com/blog/products/identity-security/enable-keyless-access-to-gcp-with-workload-identity-federation

Proposed solution: In logstash-output-google_pubsub plugin, the pubsub client is defined in logstash-output-google_pubsub/lib/logstash/outputs/pubsub/client.rb file, where the following code block (line 51 to 69 in the original file) shows how the service key is used to authenticate with pubsub.

        def initialize_google_client(json_key_file, topic_name, batch_settings)
          @logger.info("Initializing Google API client on #{topic_name} key: #{json_key_file}")

          if use_default_credential? json_key_file
            credentials = com.google.cloud.pubsub.v1.TopicAdminSettings.defaultCredentialsProviderBuilder().build()
          else
            raise_key_file_error(json_key_file)

            key_file = java.io.FileInputStream.new(json_key_file)
            sac = com.google.auth.oauth2.ServiceAccountCredentials.fromStream(key_file)
            credentials = com.google.api.gax.core.FixedCredentialsProvider.create(sac)
          end

          com.google.cloud.pubsub.v1.Publisher.newBuilder(topic_name)
             .setCredentialsProvider(credentials)
             .setHeaderProvider(construct_headers)
             .setBatchingSettings(batch_settings)
             .build
        end

A possible solution would be to replace ServiceAccountCredentials (https://googleapis.dev/java/google-auth-library/latest/com/google/auth/oauth2/ServiceAccountCredentials.html) class with ExternalAccountCredentials (https://googleapis.dev/java/google-auth-library/latest/com/google/auth/oauth2/ExternalAccountCredentials.html). ExternalAccountCredentials class will use a credentials.json file generated from workload identity federation to access gcp pubsub.

ExternalAccountCredentials credentials = 
    ExternalAccountCredentials.fromStream(new FileInputStream("/path/to/credentials.json"));

I am happy to make this contribution if everyone agrees on the proposed solution.