Getting started

Notes

• By default, TunnelWorkerConfig starts thread pools to read and process data. If you use a single machine and need to start multiple TunnelWorker instances, share one TunnelWorkerConfig.

• TunnelWorker initialization requires a ramp-up period. This period is affected by the `HeartbeatInterval` parameter in TunnelWorkerConfig. The default value is 30 s.

• If a client (TunnelWorker) does not shut down as expected, for example, due to an unexpected exit or manual termination, TunnelWorker automatically recycles resources. This includes releasing the thread pool, calling the shutdown method registered by the user on the channel, and closing the tunnel connection.

• The retention period for a tunnel's incremental logs is the same as the log time-to-live (TTL) of the stream in the data table. Because the maximum TTL is 7 days, a tunnel's incremental logs are retained for a maximum of 7 days.

• When you consume data from an incremental tunnel or a full and incremental tunnel, the following situations may occur:

  • During the full data phase, if the full data is not consumed within the incremental log retention period (a maximum of 7 days), an OTSTunnelExpired error occurs. This action prevents you from consuming subsequent data.

    If you anticipate that the full data cannot be consumed within the specified period, contact Tablestore technical support or join DingTalk group 23307953 (Tablestore Technical Exchange Group-2) for assistance.

  • During the incremental data phase, if the incremental data is not consumed within the retention period (a maximum of 7 days), the tunnel may start consumption from the most recent available data. This creates a risk of data loss.

• After a tunnel expires, Tablestore may disable it. If the tunnel remains disabled for more than 30 days, it is permanently deleted and cannot be recovered.

Prerequisites

• A data table is created. For more information, see Create a data table in the console, Create a data table using the command line interface, or Create a data table using an SDK.

• The instance endpoint is obtained. For more information, see Obtain an instance endpoint.

• Access credentials are configured. For more information, see Configure access credentials.

Try out Tunnel Service

1. Initialize the Tunnel client.

   When you initialize the Tunnel client, you can use a long-term access credential or a temporary access credential for signature authentication.

   • Initialize the client using a long-term access credential

     Before you run this code sample, ensure that the TABLESTORE_ACCESS_KEY_ID and TABLESTORE_ACCESS_KEY_SECRET environment variables are set. These variables correspond to the AccessKey ID and AccessKey secret of your Alibaba Cloud account or Resource Access Management (RAM) user.

     Warning

     The AccessKey of an Alibaba Cloud account grants full permissions on all resources. A leaked AccessKey poses a significant security threat to your system. For security reasons, we recommend that you use the AccessKey of a RAM user with the minimum required permissions instead.

     // The endpoint of the Tablestore instance. For example, https://instance.cn-hangzhou.ots.aliyun.com.
     // The name of the instance.
     // The AccessKey ID and AccessKey secret of the Alibaba Cloud account or RAM user.
     endpoint := "yourEndpoint"
     instance := "yourInstance"
     accessKeyId := os.Getenv("TABLESTORE_ACCESS_KEY_ID")
     accessKeySecret := os.Getenv("TABLESTORE_ACCESS_KEY_SECRET")
     tunnelClient := tunnel.NewTunnelClient(endpoint, instance, accessKeyId, accessKeySecret)                    

   • Initialize the client using a temporary access credential

     1. If you need temporary access to Tablestore using the Go SDK, you can use Security Token Service (STS) to issue a temporary access credential. For more information, see Temporary access credentials.

     2. The Tunnel Client provides the NewTunnelClientWithToken interface to initialize the Tunnel Client with a temporary access credential. This topic provides a code sample to help you use this interface. The sample includes logic for refreshing the temporary credential. For the complete code, see Appendix: Sample code for initializing the Tunnel Client with a temporary access credential.

2. Create a channel.

   req := &tunnel.CreateTunnelRequest{
      TableName:  "<TABLE_NAME>",
      TunnelName: "<TUNNEL_NAME>",
      Type:       tunnel.TunnelTypeBaseStream, // Create a tunnel of the full plus incremental type.
   }
   resp, err := tunnelClient.CreateTunnel(req)
   if err != nil {
      log.Fatal("create test tunnel failed", err)
   }
   log.Println("tunnel id is", resp.TunnelId)

3. Define a custom data consumption callback function as needed and start automatic data consumption.

   // Define a custom data consumption callback function based on your business needs.
   func exampleConsumeFunction(ctx *tunnel.ChannelContext, records []*tunnel.Record) error {
       fmt.Println("user-defined information", ctx.CustomValue)
       for _, rec := range records {
           fmt.Println("tunnel record detail:", rec.String())
       }
       fmt.Println("a round of records consumption finished")
       return nil
   }
   
   // Configure the callback in SimpleProcessFactory and configure the consumer-side TunnelWorkerConfig.
   workConfig := &tunnel.TunnelWorkerConfig{
      ProcessorFactory: &tunnel.SimpleProcessFactory{
         CustomValue: "user custom interface{} value",
         ProcessFunc: exampleConsumeFunction,
      },
   }
   
   // Use TunnelDaemon to continuously consume data from the specified tunnel.
   tunnelId := "<TUNNEL_ID>"
   daemon := tunnel.NewTunnelDaemon(tunnelClient, tunnelId, workConfig)
   log.Fatal(daemon.Run())

Appendix: Sample code for initializing the Tunnel Client with a temporary access credential

import (
    otscommon "github.com/aliyun/aliyun-tablestore-go-sdk/common"
    "github.com/aliyun/aliyun-tablestore-go-sdk/tunnel"
    "sync"
    "time"
)

type RefreshClient struct {
    lastRefresh          time.Time
    refreshIntervalInMin int
}

func NewRefreshClient(intervalInMin int) *RefreshClient {
    return &RefreshClient{
        refreshIntervalInMin: intervalInMin,
    }
}

func (c *RefreshClient) IsExpired() bool {
    now := time.Now()
    if c.lastRefresh.IsZero() || now.Sub(c.lastRefresh) > time.Duration(c.refreshIntervalInMin)*time.Minute {
        return true
    }

    return false
}

func (c *RefreshClient) Update() {
    c.lastRefresh = time.Now()
}

type clientCredentials struct {
    accessKeyID     string
    accessKeySecret string
    securityToken   string
}

func newClientCredentials(accessKeyID string, accessKeySecret string, securityToken string) *clientCredentials {
    return &clientCredentials{accessKeyID: accessKeyID, accessKeySecret: accessKeySecret, securityToken: securityToken}
}

func (c *clientCredentials) GetAccessKeyID() string {
    return c.accessKeyID
}

func (c *clientCredentials) GetAccessKeySecret() string {
    return c.accessKeySecret
}

func (c *clientCredentials) GetSecurityToken() string {
    return c.securityToken
}

type OTSCredentialsProvider struct {
    refresh *RefreshClient
    cred    *clientCredentials
    lock    sync.Mutex
}

func NewOTSCredentialsProvider() *OTSCredentialsProvider {
    return &OTSCredentialsProvider{
        // Adjust the refresh interval for the temporary access credential as needed. The interval must be shorter than the expiration time of the StsToken.
        refresh: NewRefreshClient(30),
    }
}

func (p *OTSCredentialsProvider) renewCredentials() error {
    if p.cred == nil || p.refresh.IsExpired() {
        // Obtain the user's StsToken here. Calling the AssumeRole API of RAM returns the AccessKeyId, AccessKeySecret, SecurityToken, and Expiration information.
        // After obtaining the temporary access credential, fill in the following parameters. For more information about the RAM (STS) SDK, see the RAM documentation. 
        // resp, err := GetUserOtsStsToken()
        accessKeyId := ""
        accessKeySecret := ""
        stsToken := ""
        p.cred = newClientCredentials(accessKeyId, accessKeySecret, stsToken)
        p.refresh.Update()
    }

    return nil
}

func (p *OTSCredentialsProvider) GetCredentials() otscommon.Credentials {
    p.lock.Lock()
    defer p.lock.Unlock()

    if err := p.renewCredentials(); err != nil {
        // log error
        if p.cred == nil {
            return newClientCredentials("", "", "")
        }
    }

    return p.cred
}

// NewTunnelClientWithToken is used to initialize a TunnelClient with the StsToken refresh feature.
func NewTunnelClientWithToken(endpoint, instanceName, accessId, accessKey, token string) tunnel.TunnelClient {
    return tunnel.NewTunnelClientWithToken(
        endpoint,
        instanceName,
        "",
        "",
        "",
        nil,
        tunnel.SetCredentialsProvider(NewOTSCredentialsProvider()),
    )
}