Client library connection code samples

// Create RedisURI from the MRC discovery endpoint
RedisURI redisUri = RedisURI.Builder.redis(CLUSTER_DISC_EP_ADDR, CLUSTER_DISC_EP_PORT).build();

// Configure client' resources
// Configure reconnectDelay with exponential backoff and full jitter
ClientResources resources = DefaultClientResources.builder()
                        .reconnectDelay(Delay.fullJitter(
    Duration.ofMillis(100),     // minimum 100 millisecond delay
    Duration.ofSeconds(5),      // maximum 5 second delay
    100, TimeUnit.MILLISECONDS) // 100 millisecond base
).build();

// Create a cluster client with the URI and resources
RedisClusterClient clusterClient = RedisClusterClient.create(resources, redisUri);

// Configure the topology refreshment options
// Enable periodic cluster topology updates so that the client updates the cluster topology in the intervals of
// 60 seconds
// Enable adaptive topology refresh that uses all triggers: MOVED_REDIRECT, ASK_REDIRECT,
// PERSISTENT_RECONNECTS, UNCOVERED_SLOT, UNKNOWN_NODE
// Disable dynamicRefreshSources so that only the initial seed nodes (Memorystore for Redis Cluster
// discovery endpoint) will be used as the source for topology discovery
// Enable closing stale connections when refreshing the cluster topology. This reduces the need to handle
// failed connections during command runtime.
ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                                .enablePeriodicRefresh(1, TimeUnit.MINUTES)
                                .enableAllAdaptiveRefreshTriggers()
                                .dynamicRefreshSources(false)
                                .closeStaleConnections(true)
                                .build();

// Configure the socket options
// Set connectTimeout based on your application requirements and workload
// Enable TCP keepAlive to reduce the need to handle failed connections during command runtime
SocketOptions socketOptions = SocketOptions.builder()
    .connectTimeout(CONNECT_TIMEOUT)
    .keepAlive(true)
    .build();

// Configure the client options
// Enable AutoReconnect when connection is lost
// Set nodeFilter to filter out failed nodes from the topology
// Disable validateClusterNodeMembership to allow redirecting commands to newly added nodes

clusterClient.setOptions(ClusterClientOptions.builder()
                                .topologyRefreshOptions(topologyRefreshOptions)
            .socketOptions(socketOptions)
                                .autoReconnect(true)
                                .nodeFilter(it ->
                                    ! (it.is(RedisClusterNode.NodeFlag.FAIL)
                                    || it.is(RedisClusterNode.NodeFlag.EVENTUAL_FAIL)
                                    || it.is(RedisClusterNode.NodeFlag.NOADDR)))
                                .validateClusterNodeMembership(false)
                                .build());

// Create a connection pool
GenericObjectPool<StatefulRedisClusterConnection<String, String> pool = ConnectionPoolSupport.createGenericObjectPool(() -> clusterClient.connect(), new GenericObjectPoolConfig());
pool.setMaxTotal(MAX_CONNECTIONS_IN_CONNECTION_POOL);

// Get a connection from the connection pool
StatefulRedisClusterConnection<String, String> connection = pool.borrowObject();

// Get a cluster sync command and call 'set'
RedisAdvancedClusterCommands<String, String> syncCommands = connection.sync();
syncCommands.set(key, value);

import (
  "context"
  "crypto/tls"
  "crypto/x509"
  "io/ioutil"
  "log"
  "time"

  "github.com/go-redis/redis/v9"
)
func example() {
        // Load CA cert
        caFilePath := 
        caCert, err := ioutil.ReadFile(caFilePath)
        if err != nil {
                log.Fatal(err)
        }
        caCertPool := x509.NewCertPool()
        caCertPool.AppendCertsFromPEM(caCert)

        // Setup Redis Connection pool
        client := redis.NewClusterClient(&redis.ClusterOptions{
                Addrs: []string{"CLUSTER_DISC_EP_ADDR:CLUSTER_DISC_EP_PORT"},
                // PoolSize applies per cluster node and not for the whole cluster.
                PoolSize:        10,
                ConnMaxIdleTime: 60 * time.Second,
                MinIdleConns:    1,
                TLSConfig: &tls.Config{
                        RootCAs: caCertPool,
                },
        })

        ctx := context.Background()
        err = client.Set(ctx, "key", "value", 0).Err()
        if err != nil {
                log.Fatal(err)
        }
}

from google.cloud import iam_credentials_v1
from redis.backoff import ConstantBackoff
from redis.retry import Retry
from redis.exceptions import (
 ConnectionError,
 AuthenticationWrongNumberOfArgsError,
 AuthenticationError
)
from redis.utils import (str_if_bytes)

import redis

service_account="projects/-/serviceAccounts/<TO-DO-1: your service account that used to authenticate to Redis Cluster>"

host=<TO-DO-2: your Redis Cluster discovery endpoint ip>

ssl_ca_certs=<TO-DO-3, your trusted server ca file name>

def generate_access_token():
  # Create a client
  client = iam_credentials_v1.IAMCredentialsClient()

  # Initialize request argument(s)
  request = iam_credentials_v1.GenerateAccessTokenRequest(
      name=service_account,
      scope=['https://www.googleapis.com/auth/cloud-platform'],
  )

  # Make the request
  response = client.generate_access_token(request=request)

  # Handle the response
  return str(response.access_token)

def iam_connect(self):
  "Initialize the connection and authenticate"
  self._parser.on_connect(self)

  auth_args = (generate_access_token())
  self.send_command("AUTH", *auth_args, check_health=False)

  try:
      auth_response = self.read_response()
  except AuthenticationWrongNumberOfArgsError:
      self.send_command("AUTH", self.password, check_health=False)
      auth_response = self.read_response()

  if str_if_bytes(auth_response) != "OK":
      raise AuthenticationError("Invalid Username or Password")
# Connect to Memorystore for Redis Cluster
backoff = ConstantBackoff(3)
retry = Retry(retries=-1, backoff=backoff, supported_errors=(ConnectionError, ConnectionResetError))
r=redis.cluster.RedisCluster(host=host, port=6379,redis_connect_func=iam_connect, retry=retry, ssl=True, ssl_ca_certs=ssl_ca_certs)

print(r.get('key'))

import com.google.cloud.iam.credentials.v1.GenerateAccessTokenResponse;
import com.google.cloud.iam.credentials.v1.IamCredentialsClient;
import io.lettuce.core.RedisCredentials;
import io.lettuce.core.RedisCredentialsProvider;
import io.lettuce.core.RedisURI;
import io.lettuce.core.SocketOptions;
import io.lettuce.core.SslOptions;
import io.lettuce.core.cluster.ClusterClientOptions;
import io.lettuce.core.cluster.ClusterTopologyRefreshOptions;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
import io.lettuce.core.resource.ClientResources;
import io.lettuce.core.resource.DefaultClientResources;
import io.lettuce.core.resource.Delay;
import java.io.Closeable;
import java.io.File;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import reactor.core.publisher.Mono;

public class IAMAuth {
    /**
    * This thread-safe implementation (excluding the main app below) is intended for production use.
    * It provides a background refresh logic that shouldn't overload IAM service in case. the
    * application has many many connections (connection storms can result in IAM throttles
    * otherwise). 

    * 

    * Guidelines for implementing similar logic for other clients:

    * 1. Refresh IAM tokens in the background using a single thread/routine per client process

    * 2. Provide last error feedback inline for token retrieval to aid debugging

    * 3. Provide initial setup validation by fast-failing if the token couldn't be retrieved

    * 4. Inline getToken shouldn't execute direct IAM calls as it can overload the token retrieval
    * resulting in throttles

    * 5. Typical scale is tens of thousands of Redis connections and the IAM token is required for
    * every connection being established.

    */
    private static final class RedisClusterCredentialsProvider
    implements RedisCredentialsProvider, Runnable, Closeable {
        private static final Logger logger =
        Logger.getLogger(RedisClusterCredentialsProvider.class.getName());

        private final ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();

        private final IamCredentialsClient iamClient;
        private final String accountName;
        private final Duration refreshDuration;
        private final Duration lifetime;

        private volatile RedisCredentials credentials;
        private volatile Instant lastRefreshInstant;
        private volatile Exception lastException;

        /**
        * AccountName:
        * "projects/-/serviceAccounts/example-service-account@example-project.iam.gserviceaccount.com";
        * RefreshDuration: Duration.ofSeconds(300) Lifetime: Duration.ofSeconds(3600);
        */
        public RedisClusterCredentialsProvider(
        String accountName, Duration refreshDuration, Duration lifetime) throws Exception {
            this.iamClient = IamCredentialsClient.create();
            this.accountName = accountName;
            this.refreshDuration = refreshDuration;
            this.lifetime = lifetime;

            // execute on initialization to fast-fail if there are any setup issues
            refreshTokenNow();
            // refresh much more frequently than the expiry time to allow for multiple retries in case of
            // failures
            service.scheduleWithFixedDelay(this, 10, 10, TimeUnit.SECONDS);
        }

        @Override
        public Mono resolveCredentials() {
            if (hasTokenExpired()) {
                throw new RuntimeException("Background IAM token refresh failed", lastException);
            }
            return Mono.just(this.credentials);
        }

        private boolean hasTokenExpired() {
            if (this.lastRefreshInstant == null || this.lifetime == null) {
                return true;
            }
            return Instant.now().isAfter(this.lastRefreshInstant.plus(this.lifetime));
        }

        // To be invoked by customer app on shutdown
        @Override
        public void close() {
            service.shutdown();
            iamClient.close();
        }

        @Override
        public void run() {
            try {
                // fetch token if it is time to refresh
                if (this.lastRefreshInstant != null
                    && this.refreshDuration != null
                && Instant.now().isBefore(this.lastRefreshInstant.plus(this.refreshDuration))) {
                    // nothing to do
                    return;
                }
                refreshTokenNow();
            } catch (Exception e) {
                // suppress all errors as we cannot allow the task to die
                // log for visibility
                logger.log(Level.parse("SEVERE"), "Background IAM token refresh failed", e);
            }
        }

        private void refreshTokenNow() {
            try {
                logger.info("Refreshing IAM token");
                com.google.protobuf.Duration lifetimeProto =
                com.google.protobuf.Duration.newBuilder()
                .setSeconds(lifetime.getSeconds())
                .setNanos(lifetime.getNano())
                .build();
                GenerateAccessTokenResponse response =
                this.iamClient.generateAccessToken(
                this.accountName,
                new ArrayList<>(),
                Collections.singletonList("https://www.googleapis.com/auth/cloud-platform"),
                lifetimeProto);

                // got a successful token refresh
                this.credentials =
                new RedisCredentials() {

                    @Override
                    public boolean hasUsername() {
                        return false;
                    }

                    @Override
                    public boolean hasPassword() {
                        return true;
                    }

                    @Override
                    public String getUsername() {
                        return "default";
                    }

                    @Override
                    public char[] getPassword() {
                        return response.getAccessToken().toCharArray();
                    }
                };
                this.lastRefreshInstant = Instant.now();
                // clear the last saved exception
                this.lastException = null;
                logger.info(
                "IAM token refreshed with lastRefreshInstant ["
                + lastRefreshInstant
                + "], refreshDuration ["
                + this.refreshDuration
                + "], accountName ["
                + this.accountName
                + "] and lifetime ["
                + this.lifetime
                + "]");
            } catch (Exception e) {
                // Save last exception for inline feedback
                this.lastException = e;
                // Bubble up for direct feedback
                throw e;
            }
        }
    }

    /** Sample code to demonstrate how to use IAMAuth; not intended for production use */
    public static void main(String[] args) throws Exception {
        // These are the parameters the user needs to replace
        String discoveryEndpointIp = "CLUSTER_DISCOVERY_ENDPOINT_IP_ADDRESS";
        int discoveryEndpointPort = CLUSTER_DISCOVERY_ENDPOINT_PORT_NUMBER;
        String accountName = "ACCOUNT_NAME";
        String caFileName = "CA_FILE_NAME";
        int refreshDurationSec = REFRESH_DURATION_SEC;
        int lifetimeSec = LIFETIME_SEC;

        RedisCredentialsProvider credentialsProvider =
        new RedisClusterCredentialsProvider(
        accountName, Duration.ofSeconds(refreshDurationSec), Duration.ofSeconds(lifetimeSec));

        RedisURI redisUri =
        RedisURI.Builder.redis(discoveryEndpointIp, discoveryEndpointPort)
        .withSsl(true)
        .withAuthentication(credentialsProvider)
        .build();
        ClientResources resources =
        DefaultClientResources.builder()
        .reconnectDelay(
        Delay.fullJitter(
        Duration.ofMillis(100), Duration.ofSeconds(5), 100, TimeUnit.MILLISECONDS))
        .build();
        ClusterTopologyRefreshOptions topologyRefreshOptions =
        ClusterTopologyRefreshOptions.builder()
        .enablePeriodicRefresh(1, TimeUnit.MINUTES)
        .enableAllAdaptiveRefreshTriggers()
        .dynamicRefreshSources(false)
        .closeStaleConnections(true)
        .build();

        SslOptions sslOptions =
        SslOptions.builder().jdkSslProvider().trustManager(new File(caFileName)).build();

        SocketOptions socketOptions =
        SocketOptions.builder().connectTimeout(Duration.ofSeconds(5)).keepAlive(true).build();

        // Create Redis Cluster Client
        RedisClusterClient clusterClient = RedisClusterClient.create(resources, redisUri);
        clusterClient.setOptions(
        ClusterClientOptions.builder()
        .topologyRefreshOptions(topologyRefreshOptions)
        .socketOptions(socketOptions)
        .sslOptions(sslOptions)
        .autoReconnect(true)
        .nodeFilter(
        it ->
        !(it.is(RedisClusterNode.NodeFlag.FAIL)
        || it.is(RedisClusterNode.NodeFlag.EVENTUAL_FAIL)
        || it.is(RedisClusterNode.NodeFlag.NOADDR)))
        .validateClusterNodeMembership(false)
        .build());

        // Establish connection to Redis Cluster
        StatefulRedisClusterConnection<String, String> connection = clusterClient.connect();

        // Retrieve synchronous Redis Cluster commands
        RedisAdvancedClusterCommands<String, String> syncCommands = connection.sync();

        // Perform Redis operations
        syncCommands.set("key1", "value1");
        String value = syncCommands.get("key1");
        System.out.println("Retrieved value: " + value);

        int count = 0;
        for (int i = 0; i < 1000; i++) {
            String k = "lettucekey" + String.valueOf(i);
            String v = "lettucevalue" + String.valueOf(i);
            syncCommands.set(k, v);
            String got = syncCommands.get(k);
            if (got.equals(v)) {
                count++;
            } else {
                System.out.println("unexpected value");
            }
        }
        System.out.println("Successfully got " + String.valueOf(count) + " keys");

        // Close the connection and shutdown the client
        connection.close();
        clusterClient.shutdown();
        ((Closeable) credentialsProvider).close();
    }
}

import com.google.cloud.iam.credentials.v1.GenerateAccessTokenResponse;
import com.google.cloud.iam.credentials.v1.IamCredentialsClient;
import java.io.Closeable;
import java.io.FileInputStream;
import java.io.InputStream;
import java.security.KeyStore;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.Connection;
import redis.clients.jedis.DefaultJedisClientConfig;
import redis.clients.jedis.DefaultRedisCredentials;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.RedisCredentials;
import redis.clients.jedis.RedisCredentialsProvider;

/** Customers are free to update/replace code as they see fit. */
public class IAMAuth {

    /**
    * This thread-safe implementation (excluding the main app below) is intended for production use.
    * It provides a background refresh logic that shouldn't overload IAM service in case. the
    * application has many many connections (connection storms can result in IAM throttles
    * otherwise). 

    * 

    * Guidelines for implementing similar logic for other clients:

    * 1. Refresh IAM tokens in the background using a single thread/routine per client process

    * 2. Provide last error feedback inline for token retrieval to aid debugging

    * 3. Provide initial setup validation by fast-failing if the token couldn't be retrieved

    * 4. Inline getToken shouldn't execute direct IAM calls as it can overload the token retrieval
    * resulting in throttles

    * 5. Typical scale is tens of thousands of Redis connections and the IAM token is required for
    * every connection being established.

    */
    private static final class RedisClusterCredentialsProvider
    implements RedisCredentialsProvider, Runnable, Closeable {
        private static final Logger logger =
        Logger.getLogger(RedisClusterCredentialsProvider.class.getName());

        private final IamCredentialsClient iamClient;
        private final ScheduledExecutorService service;
        private final String accountName;
        private final Duration refreshDuration;
        private final Duration lifetime;

        private volatile RedisCredentials credentials;
        private volatile Instant lastRefreshInstant;
        private volatile Exception lastException;

        // AccountName:
        // "projects/-/serviceAccounts/example-service-account@example-project.iam.gserviceaccount.com";
        // RefreshDuration: Duration.ofSeconds(300);
        // Lifetime: Duration.ofSeconds(3600);
        public RedisClusterCredentialsProvider(
        String accountName, Duration refreshDuration, Duration lifetime) throws Exception {
            this.iamClient = IamCredentialsClient.create();
            this.service = Executors.newSingleThreadScheduledExecutor();

            this.accountName = accountName;
            this.refreshDuration = refreshDuration;
            this.lifetime = lifetime;

            // execute on initialization to fast-fail if there are any setup issues
            refreshTokenNow();
            // refresh much more frequently than the expiry time to allow for multiple retries in case of
            // failures
            service.scheduleWithFixedDelay(this, 10, 10, TimeUnit.SECONDS);
        }

        public RedisCredentials get() {
            if (hasTokenExpired()) {
                throw new RuntimeException("Background IAM token refresh failed", lastException);
            }
            return this.credentials;
        }

        private boolean hasTokenExpired() {
            if (this.lastRefreshInstant == null || this.lifetime == null) {
                return true;
            }
            return Instant.now().isAfter(this.lastRefreshInstant.plus(this.lifetime));
        }

        // To be invoked by customer app on shutdown
        @Override
        public void close() {
            service.shutdown();
            iamClient.close();
        }

        @Override
        public void run() {
            try {
                // fetch token if it is time to refresh
                if (this.lastRefreshInstant != null
                    && this.refreshDuration != null
                && Instant.now().isBefore(this.lastRefreshInstant.plus(this.refreshDuration))) {
                    // nothing to do
                    return;
                }
                refreshTokenNow();
            } catch (Exception e) {
                // suppress all errors as we cannot allow the task to die
                // log for visibility
                logger.log(Level.parse("SEVERE"), "Background IAM token refresh failed", e);
            }
        }

        private void refreshTokenNow() {
            try {
                logger.info("Refreshing IAM token");
                List delegates = new ArrayList<>();
                com.google.protobuf.Duration lifetimeProto =
                com.google.protobuf.Duration.newBuilder()
                .setSeconds(lifetime.getSeconds())
                .setNanos(lifetime.getNano())
                .build();
                GenerateAccessTokenResponse response =
                iamClient.generateAccessToken(
                this.accountName,
                delegates,
                Collections.singletonList("https://www.googleapis.com/auth/cloud-platform"),
                lifetimeProto);

                // got a successful token refresh
                this.credentials = new DefaultRedisCredentials("default", response.getAccessToken());
                this.lastRefreshInstant = Instant.now();
                // clear the last saved exception
                this.lastException = null;
                logger.info(
                "IAM token refreshed with lastRefreshInstant ["
                + lastRefreshInstant
                + "], refreshDuration ["
                + this.refreshDuration
                + "], accountName ["
                + this.accountName
                + "] and lifetime ["
                + this.lifetime
                + "]");
            } catch (Exception e) {
                // Save last exception for inline feedback
                this.lastException = e;
                // Bubble up for direct feedback
                throw e;
            }
        }
    }

    /** Sample code to demonstrate how to use IAMAuth; not intended for production use */
    public static void main(String[] args) throws Exception {
        String discoveryEndpointIp = "CLUSTER_DISCOVERY_ENDPOINT_IP_ADDRESS";
        int discoveryEndpointPort = CLUSTER_DISCOVERY_ENDPOINT_PORT_NUMBER;

        GenericObjectPoolConfig config = new GenericObjectPoolConfig();
        config.setTestWhileIdle(true);
        int timeout = 5000;
        int maxAttempts = 5;
        HostAndPort discovery = new HostAndPort(discoveryEndpointIp, discoveryEndpointPort);

        RedisCredentialsProvider credentialsProvider =
        new RedisClusterCredentialsProvider(
        "projects/-/serviceAccounts/example-service-account@example-project.iam.gserviceaccount.com",
        Duration.ofSeconds(300),
        Duration.ofSeconds(3600));

        // Create JedisCluster cluster
        InputStream is = new FileInputStream("server-ca.pem");
        // You could get a resource as a stream instead.

        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        X509Certificate caCert = (X509Certificate) cf.generateCertificate(is);

        TrustManagerFactory tmf =
        TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
        ks.load(null); // You don't need the KeyStore cluster to come from a file.
        ks.setCertificateEntry("caCert", caCert);

        tmf.init(ks);

        SSLContext sslContext = SSLContext.getInstance("TLS");
        sslContext.init(null, tmf.getTrustManagers(), null);

        JedisCluster jedisCluster =
        new JedisCluster(
        discovery,
        DefaultJedisClientConfig.builder()
        .connectionTimeoutMillis(timeout)
        .socketTimeoutMillis(timeout)
        .credentialsProvider(credentialsProvider)
        .ssl(true)
        .sslSocketFactory(sslContext.getSocketFactory())
        .build(),
        maxAttempts,
        config);

        // Perform operations on the cluster
        jedisCluster.set("myKey", "Hello, Redis Cluster!");
        String value = jedisCluster.get("myKey");
        System.out.println("Value for myKey: " + value);

        int count = 0;
        for (int i = 0; i < 1000; i++) {
            String k = "jediskey" + String.valueOf(i);
            String v = "jedisvalue" + String.valueOf(i);
            jedisCluster.set(k, v);
            String got = jedisCluster.get(k);
            if (got.equals(v)) {
                count++;
            } else {
                System.out.println("unexpected value");
            }
        }
        System.out.println("Successfully got " + String.valueOf(count) + " keys");

        // Disconnect from the cluster
        jedisCluster.close();
        // Cleanup the resources used by the provider
        ((Closeable) credentialsProvider).close();
    }
}

package main

import (
    "context"
    "crypto/tls"
    "crypto/x509"
    "flag"
    "fmt"
    "io/ioutil"
    "log"
    "sync"
    "time"

    credentials "google.golang.org/genproto/googleapis/iam/credentials/v1"

    "github.com/golang/protobuf/ptypes"
    "github.com/redis/go-redis/v9"
    "google.golang.org/api/option"
    gtransport "google.golang.org/api/transport/grpc"
)

var (
    svcAccount               = flag.String("a", "projects/-/serviceAccounts/example-service-account@example-project.iam.gserviceaccount.com", "service account email")
    lifetime                 = flag.Duration("d", time.Hour, "lifetime of token")
    refreshDuration          = flag.Duration("r", 5*time.Minute, "token refresh duration")
    checkTokenExpiryInterval = flag.Duration("e", 10*time.Second, "check token expiry interval")
    lastRefreshInstant       = time.Time{}
    errLastSeen              = error(nil)
    token                    = ""
    mu                       = sync.RWMutex{}
)

func retrieveToken() (string, error) {
    ctx := context.Background()
    conn, err := gtransport.Dial(ctx,
        option.WithEndpoint("iamcredentials.googleapis.com:443"),
        option.WithScopes("https://www.googleapis.com/auth/cloud-platform"))

    if err != nil {
        log.Printf("Failed to dial API, error: %v", err)
        return token, err
    }
    client := credentials.NewIAMCredentialsClient(conn)
    req := credentials.GenerateAccessTokenRequest{
        Name:     *svcAccount,
        Scope:    []string{"https://www.googleapis.com/auth/cloud-platform"},
        Lifetime: ptypes.DurationProto(*lifetime),
    }
    rsp, err := client.GenerateAccessToken(ctx, &req)
    if err != nil {
        log.Printf"Failed to call GenerateAccessToken with request: %v, error: %v", req, err)
        return token, err
    }
    return rsp.AccessToken, nil
}

func refreshTokenLoop() {
    if *refreshDuration > *lifetime {
        log.Fatal("Refresh should not happen after token is already expired.")
    }
    for {
        mu.RLock()
        lastRefreshTime := lastRefreshInstant
        mu.RUnlock()
        if time.Now().After(lastRefreshTime.Add(*refreshDuration)) {
            var err error
            retrievedToken, err := retrieveToken()
            mu.Lock()
            token = retrievedToken
            if err != nil {
                errLastSeen = err
            } else {
                lastRefreshInstant = time.Now()
            }
            mu.Unlock()
        }
        time.Sleep(*checkTokenExpiryInterval)
    }
}

func retrieveTokenFunc() (string, string) {
    mu.RLock()
    defer mu.RUnlock()
    if time.Now().After(lastRefreshInstant.Add(*refreshDuration)) {
        log.Printf("Token is expired. last refresh instant: %v, refresh duration: %v, error that was last seen: %v", lastRefreshInstant, *refreshDuration, errLastSeen)
        return "", ""
    }
    username := "default"
    password := token
    return username, password
}

func main() {
    // Load CA cert
    caFilePath := CA_FILE_PATH
    clusterDicEpAddr := CLUSTER_DISCOVERY_ENDPOINT_IP_ADDRESS_AND_PORT
    caCert, err := ioutil.ReadFile(caFilePath)
    if err != nil {
        log.Fatal(err)
    }
    caCertPool := x509.NewCertPool()
    caCertPool.AppendCertsFromPEM(caCert)
    token, err = retrieveToken()
    if err != nil {
        log.Fatal("Cannot retrieve IAM token to authenticate to the cluster, error: %v", err)
    }
    lastRefreshInstant = time.Now()
    go refreshTokenLoop()

    // Setup Redis Connection pool
    client := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs: []string{clusterDicEpAddr},
        // PoolSize applies per cluster node and not for the whole cluster.
        PoolSize:            10,
        ConnMaxIdleTime:     60 * time.Second,
        MinIdleConns:        1,
        CredentialsProvider: retrieveTokenFunc,
        TLSConfig: &tls.Config{
            RootCAs: caCertPool,
        },
    })

    ctx := context.Background()
    err = client.Set(ctx, "key", "value", 0).Err()
    if err != nil {
        log.Fatal(err)
    }
    val, err := client.Get(ctx, "key").Result()
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Got the value for key: key, which is %s \n", val)
}