Introducción - Tablestore - Centro de documentación de Alibaba Cloud

Tunnel Service le permite consumir los datos de una tabla. Este tema describe cómo comenzar a utilizar Tunnel Service mediante Tablestore SDK for Go. Antes de usar Tunnel Service, asegúrese de conocer las notas de uso del servicio.

Notas de uso

De forma predeterminada, el sistema inicia un pool de hilos para leer y procesar datos según TunnelWorkerConfig. Si desea iniciar varios TunnelWorkers en un solo servidor, se recomienda usar la misma configuración de TunnelWorkerConfig para todos ellos.
TunnelWorker requiere un periodo de calentamiento para su inicialización, especificado mediante el parámetro HeartbeatInterval en TunnelWorkerConfig. Valor predeterminado: 30. Unidad: segundos.
Cuando el cliente de TunnelWorker se detiene debido a una salida inesperada o a una finalización manual, TunnelWorker recicla automáticamente los recursos mediante uno de los siguientes métodos: libera el pool de hilos, invoca automáticamente el método de cierre que registró para la clase Channel y cierra el túnel.
El periodo de retención de los registros incrementales en los túneles es igual al de los registros de Stream. Los registros de Stream pueden conservarse hasta por siete días; por lo tanto, los registros incrementales en los túneles también tienen un límite de retención de siete días.
Si crea un túnel para consumir datos diferenciales o incrementales, tenga en cuenta lo siguiente:
- Durante el consumo completo de datos, si el túnel no logra finalizar dicho consumo dentro del periodo de retención de los registros incrementales (siete días como máximo), se produce un error OTSTunnelExpired cuando el túnel comienza a consumir los registros incrementales. Como resultado, el túnel no puede consumir dichos registros.
  
  Si estima que el túnel no podrá completar el consumo total de datos dentro de la ventana de tiempo especificada, póngase en contacto con el soporte técnico de Tablestore .
- Durante el consumo de datos incrementales, si el túnel no logra completar el consumo de los registros incrementales dentro de su periodo de retención (siete días como máximo), podría consumir datos a partir del punto más reciente disponible. En este caso, es posible que algunos datos específicos no se consuman.
Una vez que un túnel caduca, Tablestore puede deshabilitarlo. Si un túnel permanece en estado deshabilitado durante más de 30 días, se elimina. No es posible restaurar un túnel eliminado.

Requisitos previos

Se ha creado una tabla de datos. Para obtener más información, consulte Use the Tablestore console, Use the Tablestore CLI y Use Tablestore SDKs.
Se ha obtenido el endpoint de la instancia donde reside la tabla de datos. Para obtener más información, consulte Obtain an endpoint of a Tablestore instance.
Se han configurado las credenciales de acceso. Para obtener más información, consulte Configure access credentials.

Comenzar a utilizar Tunnel Service

Inicialice una instancia de TunnelClient.

Al inicializar una instancia de TunnelClient, puede utilizar credenciales de acceso permanentes o temporales para la autenticación.
- Uso de credenciales de acceso permanentes para la inicialización
  
  Asegúrese de que las variables de entorno TABLESTORE_ACCESS_KEY_ID y TABLESTORE_ACCESS_KEY_SECRET estén configuradas. La variable de entorno TABLESTORE_ACCESS_KEY_ID especifica el AccessKey ID de su cuenta de Alibaba Cloud o usuario RAM. La variable de entorno TABLESTORE_ACCESS_KEY_SECRET especifica el AccessKey secret de su cuenta de Alibaba Cloud o usuario RAM.
  
  Aviso
  Una cuenta de Alibaba Cloud tiene acceso completo a todos los recursos de dicha cuenta. La filtración del par de AccessKeys de la cuenta de Alibaba Cloud representa una amenaza crítica para el sistema. Por lo tanto, se recomienda utilizar el par de AccessKeys de un usuario RAM al que se le hayan otorgado únicamente los permisos mínimos obligatorios para inicializar una instancia de TunnelClient.
```
// Set the endpoint parameter to the endpoint of the Tablestore instance. Example: https://instance.cn-hangzhou.ots.aliyuncs.com. 
// Specify the name of the instance. 
// Specify the AccessKey ID and AccessKey secret of your Alibaba Cloud account or a 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)                    
```
- Uso de credenciales de acceso temporales para la inicialización
  1. Si desea acceder temporalmente a Tablestore mediante Tablestore SDK for Go, puede utilizar Security Token Service (STS) para generar credenciales de acceso temporales. Para obtener más información, consulte Configure temporary access credentials.
  2. Un cliente de túnel proporciona la operación NewTunnelClientWithToken, que puede invocar para inicializar una instancia de TunnelClient basada en credenciales de acceso temporales. Este tema proporciona código de ejemplo para inicializar una instancia de TunnelClient mediante credenciales de acceso temporales que pueden actualizarse periódicamente. Para obtener más información, consulte Appendix: Sample code for initializing a TunnelClient instance by using temporary access credentials.

Cree un túnel.

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

Especifique una función de callback personalizada para iniciar el consumo automático de datos.

// Specify a custom callback function. 
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 function. Information about the callback function is passed to SimpleProcessFactory. Configure TunnelWorkerConfig for the consumer. 
workConfig := &tunnel.TunnelWorkerConfig{
   ProcessorFactory: &tunnel.SimpleProcessFactory{
      CustomValue: "user custom interface{} value",
      ProcessFunc: exampleConsumeFunction,
   },
}

// Use TunnelDaemon to continuously consume the specified tunnel. 
tunnelId := "<TUNNEL_ID>"
daemon := tunnel.NewTunnelDaemon(tunnelClient, tunnelId, workConfig)
log.Fatal(daemon.Run())

Apéndice: Código de ejemplo para inicializar una instancia de TunnelClient mediante credenciales de acceso temporales

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{
        // Modify the refresh cycle for temporary access credentials based on your business requirements. The refresh cycle must be shorter than the validity period of the temporary access credentials. 
        refresh: NewRefreshClient(30),
    }
}

func (p *OTSCredentialsProvider) renewCredentials() error {
    if p.cred == nil || p.refresh.IsExpired() {
        // Obtain temporary access credentials. You can call the AssumeRole operation of RAM to obtain the AccessKey ID, AccessKey secret, security token, and validity period of the temporary access credentials. 
        // Configure the following parameters. For information about RAM SDKs, see the documentation of RAM.  
        // 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 instance with the feature of refreshing temporary access credentials. 
func NewTunnelClientWithToken(endpoint, instanceName, accessId, accessKey, token string) tunnel.TunnelClient {
    return tunnel.NewTunnelClientWithToken(
        endpoint,
        instanceName,
        "",
        "",
        "",
        nil,
        tunnel.SetCredentialsProvider(NewOTSCredentialsProvider()),
    )
}