Tablestore:Primeiros passos

Notas de uso

• Por padrão, o sistema inicia um pool de threads para ler e processar dados com base no TunnelWorkerConfig. Para iniciar vários TunnelWorkers em um único servidor, use o mesmo TunnelWorkerConfig na configuração de todos eles.

• O TunnelWorker exige um período de aquecimento para inicialização, definido pelo parâmetro HeartbeatInterval no TunnelWorkerConfig. Valor padrão: 30. Unidade: segundos.

• Se o cliente TunnelWorker for encerrado devido a uma saída inesperada ou término manual, ele reciclará automaticamente os recursos por meio de um dos seguintes métodos: liberação do pool de threads, chamada automática do método de encerramento registrado para a classe Channel e fechamento do túnel.

• O período de retenção de logs incrementais nos túneis corresponde ao período de retenção de logs do Stream. Como os logs do Stream podem ser retidos por até sete dias, os logs incrementais nos túneis também têm retenção máxima de sete dias.

• Ao criar um túnel para consumir dados diferenciais ou incrementais, observe os seguintes pontos:

  • Durante o consumo completo de dados, se o túnel não concluir o consumo dentro do período de retenção de logs incrementais (no máximo sete dias), ocorrerá um erro OTSTunnelExpired ao iniciar o consumo de logs incrementais. Consequentemente, o túnel não conseguirá consumir esses logs.

    Caso estime que o túnel não concluirá o consumo completo dos dados dentro da janela de tempo especificada, entre em contato com o suporte técnico do Tablestore .

  • Durante o consumo de dados incrementais, se o túnel não concluir o consumo dos logs incrementais dentro do período de retenção (no máximo sete dias), ele poderá consumir dados a partir dos registros disponíveis mais recentes. Nesse cenário, alguns dados específicos podem não ser consumidos.

• Após a expiração de um túnel, o Tablestore pode desativá-lo. Se permanecer no estado desativado por mais de 30 dias, o túnel será excluído. Não é possível restaurar um túnel excluído.

Pré-requisitos

• Crie uma tabela de dados. Para mais informações, consulte Usar o console do Tablestore, Usar o CLI do Tablestore e Usar SDKs do Tablestore.

• Obtenha o endpoint da instância onde a tabela de dados está localizada. Para mais informações, consulte Obter um endpoint de uma instância do Tablestore.

• Configure as credenciais de acesso. Para mais informações, consulte Configurar credenciais de acesso.

Começar a usar o Tunnel Service

1. Inicialize uma instância TunnelClient.

   Na inicialização de uma instância TunnelClient, use credenciais de acesso de longo prazo ou temporárias para autenticação.

   • Use credenciais de acesso de longo prazo para inicialização

     Certifique-se de que as variáveis de ambiente TABLESTORE_ACCESS_KEY_ID e TABLESTORE_ACCESS_KEY_SECRET estejam configuradas. A variável TABLESTORE_ACCESS_KEY_ID especifica o AccessKey ID da sua conta Alibaba Cloud ou usuário RAM. A variável TABLESTORE_ACCESS_KEY_SECRET especifica o AccessKey secret da sua conta Alibaba Cloud ou usuário RAM.

     Aviso

     Uma conta Alibaba Cloud tem acesso total a todos os recursos da conta. O vazamento do par de AccessKeys da conta Alibaba Cloud representa ameaças críticas ao sistema. Portanto, recomendamos usar o par de AccessKeys de um usuário RAM com as permissões mínimas necessárias para inicializar uma instância 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)                    

   • Use credenciais de acesso temporárias para inicialização

     1. Para acessar o Tablestore temporariamente com o Tablestore SDK for Go, use o Security Token Service (STS) para gerar credenciais de acesso temporárias. Para mais informações, consulte Configurar credenciais de acesso temporárias.

     2. O cliente de túnel fornece a operação NewTunnelClientWithToken para inicializar uma instância TunnelClient com base em credenciais de acesso temporárias. Este tópico fornece um código de exemplo para inicializar uma instância TunnelClient usando credenciais de acesso temporárias atualizáveis periodicamente. Para mais informações, consulte Apêndice: Código de exemplo para inicializar uma instância TunnelClient usando credenciais de acesso temporárias.

2. Crie um 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)

3. Especifique uma função de callback personalizada para iniciar o consumo automático de dados.

   // 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 exemplo para inicializar uma instância TunnelClient usando credenciais de acesso temporárias

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()),
    )
}