Tutorial 1: Criar um serviço Snowpark Container Services¶

Pré-requisitos¶

Certifique-se de ter as seguintes informações:

• URL de entrada do ponto de extremidade público. Execute o comando SHOW ENDPOINTS IN SERVICEpara obter o URL:

  SHOW ENDPOINTS IN SERVICE echo_service;
  

  Copy

• Nome de sua conta Snowflake. Para obter mais informações, consulte Configuração comum: Verifique se você está pronto para continuar.

• URL de sua conta Snowflake: É <acctname >.snowflakecomputing.com.

• Nome de usuário na conta Snowflake. Este é o usuário escolhido em Configuração comum: Criação de objetos Snowflake. Você faz login no Snowflake como este usuário e testa o acesso programático.

• Nome da função: você criou uma função (test_role) como parte da configuração comum. O usuário assume essa função para executar ações.

Configuração¶

Siga as etapas para se comunicar com o serviço Echo programaticamente. Como usar o código Python fornecido, você envia solicitações ao ponto de extremidade público que o serviço Echo expõe.

1. Em um prompt de comando, crie um diretório e navegue até ele.

2. Configure a autenticação do par de chaves para o usuário.

   1. Gerar um par de chaves:

      1. Gere uma chave privada. Para simplificar as etapas do exercício, você gera uma chave privada não criptografada. Você também pode usar uma chave privada criptografada, mas será necessário digitar a senha.

         openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
         

         Copy

      2. Gere uma chave pública (rsa_key.pub) referenciando a chave privada que você criou.

         openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
         

         Copy

   2. Verifique se você tem a chave privada e a chave pública geradas no diretório.

   3. Atribua a chave pública ao usuário que você está usando para testar o acesso programático. Isso permite que o usuário especifique a chave para autenticação.

      ALTER USER <user-name> SET RSA_PUBLIC_KEY='MIIBIjANBgkqh...';
      

      Copy

3. Salve o código de amostra fornecido em arquivos Python.

   1. Salve o código a seguir em generateJWT.py.

      # To run this on the command line, enter:
      #   python3 generateJWT.py --account=<account_identifier> --user=<username> --private_key_file_path=<path_to_private_key_file>
      
      from cryptography.hazmat.primitives.serialization import load_pem_private_key
      from cryptography.hazmat.primitives.serialization import Encoding
      from cryptography.hazmat.primitives.serialization import PublicFormat
      from cryptography.hazmat.backends import default_backend
      from datetime import timedelta, timezone, datetime
      import argparse
      import base64
      from getpass import getpass
      import hashlib
      import logging
      import sys
      
      # This class relies on the PyJWT module (https://pypi.org/project/PyJWT/).
      import jwt
      
      logger = logging.getLogger(__name__)
      
      try:
          from typing import Text
      except ImportError:
          logger.debug('# Python 3.5.0 and 3.5.1 have incompatible typing modules.', exc_info=True)
          from typing_extensions import Text
      
      ISSUER = "iss"
      EXPIRE_TIME = "exp"
      ISSUE_TIME = "iat"
      SUBJECT = "sub"
      
      # If you generated an encrypted private key, implement this method to return
      # the passphrase for decrypting your private key. As an example, this function
      # prompts the user for the passphrase.
      def get_private_key_passphrase():
          return getpass('Passphrase for private key: ')
      
      class JWTGenerator(object):
          """
          Creates and signs a JWT with the specified private key file, username, and account identifier. The JWTGenerator keeps the
          generated token and only regenerates the token if a specified period of time has passed.
          """
          LIFETIME = timedelta(minutes=59)  # The tokens will have a 59-minute lifetime
          RENEWAL_DELTA = timedelta(minutes=54)  # Tokens will be renewed after 54 minutes
          ALGORITHM = "RS256"  # Tokens will be generated using RSA with SHA256
      
          def __init__(self, account: Text, user: Text, private_key_file_path: Text,
                      lifetime: timedelta = LIFETIME, renewal_delay: timedelta = RENEWAL_DELTA):
              """
              __init__ creates an object that generates JWTs for the specified user, account identifier, and private key.
              :param account: Your Snowflake account identifier. See https://docs.snowflake.com/en/user-guide/admin-account-identifier.html. Note that if you are using the account locator, exclude any region information from the account locator.
              :param user: The Snowflake username.
              :param private_key_file_path: Path to the private key file used for signing the JWTs.
              :param lifetime: The number of minutes (as a timedelta) during which the key will be valid.
              :param renewal_delay: The number of minutes (as a timedelta) from now after which the JWT generator should renew the JWT.
              """
      
              logger.info(
                  """Creating JWTGenerator with arguments
                  account : %s, user : %s, lifetime : %s, renewal_delay : %s""",
                  account, user, lifetime, renewal_delay)
      
              # Construct the fully qualified name of the user in uppercase.
              self.account = self.prepare_account_name_for_jwt(account)
              self.user = user.upper()
              self.qualified_username = self.account + "." + self.user
      
              self.lifetime = lifetime
              self.renewal_delay = renewal_delay
              self.private_key_file_path = private_key_file_path
              self.renew_time = datetime.now(timezone.utc)
              self.token = None
      
              # Load the private key from the specified file.
              with open(self.private_key_file_path, 'rb') as pem_in:
                  pemlines = pem_in.read()
                  try:
                      # Try to access the private key without a passphrase.
                      self.private_key = load_pem_private_key(pemlines, None, default_backend())
                  except TypeError:
                      # If that fails, provide the passphrase returned from get_private_key_passphrase().
                      self.private_key = load_pem_private_key(pemlines, get_private_key_passphrase().encode(), default_backend())
      
          def prepare_account_name_for_jwt(self, raw_account: Text) -> Text:
              """
              Prepare the account identifier for use in the JWT.
              For the JWT, the account identifier must not include the subdomain or any region or cloud provider information.
              :param raw_account: The specified account identifier.
              :return: The account identifier in a form that can be used to generate the JWT.
              """
              account = raw_account
              if not '.global' in account:
                  # Handle the general case.
                  idx = account.find('.')
                  if idx > 0:
                      account = account[0:idx]
              else:
                  # Handle the replication case.
                  idx = account.find('-')
                  if idx > 0:
                      account = account[0:idx]
              # Use uppercase for the account identifier.
              return account.upper()
      
          def get_token(self) -> Text:
              """
              Generates a new JWT. If a JWT has already been generated earlier, return the previously generated token unless the
              specified renewal time has passed.
              :return: the new token
              """
              now = datetime.now(timezone.utc)  # Fetch the current time
      
              # If the token has expired or doesn't exist, regenerate the token.
              if self.token is None or self.renew_time <= now:
                  logger.info("Generating a new token because the present time (%s) is later than the renewal time (%s)",
                              now, self.renew_time)
                  # Calculate the next time we need to renew the token.
                  self.renew_time = now + self.renewal_delay
      
                  # Prepare the fields for the payload.
                  # Generate the public key fingerprint for the issuer in the payload.
                  public_key_fp = self.calculate_public_key_fingerprint(self.private_key)
      
                  # Create our payload
                  payload = {
                      # Set the issuer to the fully qualified username concatenated with the public key fingerprint.
                      ISSUER: self.qualified_username + '.' + public_key_fp,
      
                      # Set the subject to the fully qualified username.
                      SUBJECT: self.qualified_username,
      
                      # Set the issue time to now.
                      ISSUE_TIME: now,
      
                      # Set the expiration time, based on the lifetime specified for this object.
                      EXPIRE_TIME: now + self.lifetime
                  }
      
                  # Regenerate the actual token
                  token = jwt.encode(payload, key=self.private_key, algorithm=JWTGenerator.ALGORITHM)
                  # If you are using a version of PyJWT prior to 2.0, jwt.encode returns a byte string instead of a string.
                  # If the token is a byte string, convert it to a string.
                  if isinstance(token, bytes):
                    token = token.decode('utf-8')
                  self.token = token
                  logger.info("Generated a JWT with the following payload: %s", jwt.decode(self.token, key=self.private_key.public_key(), algorithms=[JWTGenerator.ALGORITHM]))
      
              return self.token
      
          def calculate_public_key_fingerprint(self, private_key: Text) -> Text:
              """
              Given a private key in PEM format, return the public key fingerprint.
              :param private_key: private key string
              :return: public key fingerprint
              """
              # Get the raw bytes of public key.
              public_key_raw = private_key.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
      
              # Get the sha256 hash of the raw bytes.
              sha256hash = hashlib.sha256()
              sha256hash.update(public_key_raw)
      
              # Base64-encode the value and prepend the prefix 'SHA256:'.
              public_key_fp = 'SHA256:' + base64.b64encode(sha256hash.digest()).decode('utf-8')
              logger.info("Public key fingerprint is %s", public_key_fp)
      
              return public_key_fp
      
      def main():
          logging.basicConfig(stream=sys.stdout, level=logging.INFO)
          cli_parser = argparse.ArgumentParser()
          cli_parser.add_argument('--account', required=True, help='The account identifier (e.g. "myorganization-myaccount" for "myorganization-myaccount.snowflakecomputing.com").')
          cli_parser.add_argument('--user', required=True, help='The user name.')
          cli_parser.add_argument('--private_key_file_path', required=True, help='Path to the private key file used for signing the JWT.')
          cli_parser.add_argument('--lifetime', type=int, default=59, help='The number of minutes that the JWT should be valid for.')
          cli_parser.add_argument('--renewal_delay', type=int, default=54, help='The number of minutes before the JWT generator should produce a new JWT.')
          args = cli_parser.parse_args()
      
          token = JWTGenerator(args.account, args.user, args.private_key_file_path, timedelta(minutes=args.lifetime), timedelta(minutes=args.renewal_delay)).get_token()
          print('JWT:')
          print(token)
      
      if __name__ == "__main__":
          main()
      

      Copy

   2. Salve o código a seguir em access-via-keypair.py.

      from generateJWT import JWTGenerator
      from datetime import timedelta
      import argparse
      import logging
      import sys
      import requests
      logger = logging.getLogger(__name__)
      
      def main():
        args = _parse_args()
        token = _get_token(args)
        snowflake_jwt = token_exchange(token,endpoint=args.endpoint, role=args.role,
                        snowflake_account_url=args.snowflake_account_url,
                        snowflake_account=args.account)
        spcs_url=f'https://{args.endpoint}{args.endpoint_path}'
        connect_to_spcs(snowflake_jwt, spcs_url)
      
      def _get_token(args):
        token = JWTGenerator(args.account, args.user, args.private_key_file_path, timedelta(minutes=args.lifetime),
                  timedelta(minutes=args.renewal_delay)).get_token()
        logger.info("Key Pair JWT: %s" % token)
        return token
      
      def token_exchange(token, role, endpoint, snowflake_account_url, snowflake_account):
        scope_role = f'session:role:{role}' if role is not None else None
        scope = f'{scope_role} {endpoint}' if scope_role is not None else endpoint
        data = {
          'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
          'scope': scope,
          'assertion': token,
        }
        logger.info(data)
        url = f'https://{snowflake_account}.snowflakecomputing.com/oauth/token'
        if snowflake_account_url:
          url =       f'{snowflake_account_url}/oauth/token'
        logger.info("oauth url: %s" %url)
        response = requests.post(url, data=data)
        logger.info("snowflake jwt : %s" % response.text)
        assert 200 == response.status_code, "unable to get snowflake token"
        return response.text
      
      def connect_to_spcs(token, url):
        # Create a request to the ingress endpoint with authz.
        headers = {'Authorization': f'Snowflake Token="{token}"'}
        response = requests.post(f'{url}', headers=headers)
        logger.info("return code %s" % response.status_code)
        logger.info(response.text)
      
      def _parse_args():
        logging.basicConfig(stream=sys.stdout, level=logging.INFO)
        cli_parser = argparse.ArgumentParser()
        cli_parser.add_argument('--account', required=True,
                    help='The account identifier (for example, "myorganization-myaccount" for '
                      '"myorganization-myaccount.snowflakecomputing.com").')
        cli_parser.add_argument('--user', required=True, help='The user name.')
        cli_parser.add_argument('--private_key_file_path', required=True,
                    help='Path to the private key file used for signing the JWT.')
        cli_parser.add_argument('--lifetime', type=int, default=59,
                    help='The number of minutes that the JWT should be valid for.')
        cli_parser.add_argument('--renewal_delay', type=int, default=54,
                    help='The number of minutes before the JWT generator should produce a new JWT.')
        cli_parser.add_argument('--role',
                    help='The role we want to use to create and maintain a session for. If a role is not provided, '
                      'use the default role.')
        cli_parser.add_argument('--endpoint', required=True,
                    help='The ingress endpoint of the service')
        cli_parser.add_argument('--endpoint-path', default='/',
                    help='The url path for the ingress endpoint of the service')
        cli_parser.add_argument('--snowflake_account_url', default=None,
                    help='The account url of the account for which we want to log in. Type of '
                      'https://myorganization-myaccount.snowflakecomputing.com')
        args = cli_parser.parse_args()
        return args
      
      if __name__ == "__main__":
        main()
      

      Copy

Como enviar de solicitações ao ponto de extremidade do servidor programaticamente¶

Execute o código Python access-via-keypair.py para fazer a chamada de entrada para o ponto de extremidade público do serviço Echo.

python3 access-via-keypair.py \
  --account <account-identifier> \
  --user <user-name> \
  --role TEST_ROLE \
  --private_key_file_path rsa_key.p8 \
  --endpoint <ingress-hostname> \
  --endpoint-path /ui

Copy

Para obter mais informações sobre account-identifier, consulte Identificadores de conta.

Como funciona a autenticação¶

O código primeiro converte o par de chaves fornecido em um token JWT. Em seguida, ele envia o token JWT para o Snowflake para obter um token OAuth. Por fim, o código usa o token OAuth para se conectar ao Snowflake e acessar o ponto de extremidade público. Especificamente, o código faz o seguinte:

1. Chama a função _get_token(args) para gerar um token JWT a partir do par de chaves fornecido. A implementação da função é mostrada:

   def _get_token(args):
       token = JWTGenerator(args.account,
                           args.user,
                           args.private_key_file_path,
                           timedelta(minutes=args.lifetime),
                           timedelta(minutes=args.renewal_delay)).get_token()
       logger.info("Key Pair JWT: %s" % token)
       return token
   

   Copy

   JWTGenerator é uma classe auxiliar fornecida a você. Observe o seguinte sobre os parâmetros que você fornece ao criar este objeto:

   • Parâmetros args.account e args.user: Um token JWT tem vários campos (consulte o formato de token); iss é um dos campos. Este valor de campo inclui o nome da conta Snowflake e um nome de usuário. Portanto, você fornece esses valores como parâmetros.

   • Os dois parâmetros timedelta fornecem as seguintes informações:

     • lifetime especifica o número de minutos durante os quais a chave será válida (60 minutos).

     • renewal_delay especifica o número de minutos a partir de agora após o qual o gerador JWT deve renovar o JWT.

2. Chama a função token_exchange() para se conectar ao Snowflake e trocar o token JWT por um token OAuth.

   scope_role = f'session:role:{role}' if role is not None else None
   scope = f'{scope_role} {endpoint}' if scope_role is not None else endpoint
   
   data = {
       'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
       'scope': scope,
       'assertion': token,
   }
   

   Copy

   O código anterior constrói uma configuração JSON do escopo para o token OAuth, o ponto de extremidade público que pode ser acessado usando a função especificada. Este código então faz uma solicitação POST ao Snowflake passando o JSON para trocar o token JWT por um token OAuth (consulte Troca de tokens) conforme mostrado:

   url = f'{snowflake_account_url}/oauth/token'
   response = requests.post(url, data=data)
   assert 200 == response.status_code, "unable to get Snowflake token"
   return response.text
   

   Copy

3. O código então chama a função connect_to_spcs() para se conectar ao ponto de extremidade público do serviço Echo. Ele fornece o URL (https://<URL de entrada>/ui) do ponto de extremidade e o token OAuth para autenticação.

   headers = {'Authorization': f'Snowflake Token="{token}"'}
   response = requests.post(f'{url}', headers=headers)
   

   Copy

   O url é o spcs_url que você forneceu ao programa, e o token é o token OAuth.

   O serviço Echo neste exemplo atende uma página HTML (conforme explicado na seção anterior). Este código de amostra simplesmente imprime o HTML na resposta.

Análise do código do tutorial 1¶

O arquivo zip baixado na etapa 1 inclui os seguintes arquivos:

• Dockerfile

• echo_service.py

• templates/basic_ui.html

Você também usa a especificação de serviço ao criar o serviço. A seção a seguir explica como esses componentes de código funcionam juntos para criar o serviço.

from flask import Flask
from flask import request
from flask import make_response
from flask import render_template
import logging
import os
import sys

SERVICE_HOST = os.getenv('SERVER_HOST', '0.0.0.0')
SERVER_PORT = os.getenv('SERVER_PORT', 8080)
CHARACTER_NAME = os.getenv('CHARACTER_NAME', 'I')


def get_logger(logger_name):
  logger = logging.getLogger(logger_name)
  logger.setLevel(logging.DEBUG)
  handler = logging.StreamHandler(sys.stdout)
  handler.setLevel(logging.DEBUG)
  handler.setFormatter(
    logging.Formatter(
      '%(name)s [%(asctime)s] [%(levelname)s] %(message)s'))
  logger.addHandler(handler)
  return logger


logger = get_logger('echo-service')

app = Flask(__name__)


@app.get("/healthcheck")
def readiness_probe():
  return "I'm ready!"


@app.post("/echo")
def echo():
  '''
  Main handler for input data sent by Snowflake.
  '''
  message = request.json
  logger.debug(f'Received request: {message}')

  if message is None or not message['data']:
    logger.info('Received empty message')
    return {}

  # input format:
  #   {"data": [
  #     [row_index, column_1_value, column_2_value, ...],
  #     ...
  #   ]}
  input_rows = message['data']
  logger.info(f'Received {len(input_rows)} rows')

  # output format:
  #   {"data": [
  #     [row_index, column_1_value, column_2_value, ...}],
  #     ...
  #   ]}
  output_rows = [[row[0], get_echo_response(row[1])] for row in input_rows]
  logger.info(f'Produced {len(output_rows)} rows')

  response = make_response({"data": output_rows})
  response.headers['Content-type'] = 'application/json'
  logger.debug(f'Sending response: {response.json}')
  return response


@app.route("/ui", methods=["GET", "POST"])
def ui():
  '''
  Main handler for providing a web UI.
  '''
  if request.method == "POST":
    # getting input in HTML form
    input_text = request.form.get("input")
    # display input and output
    return render_template("basic_ui.html",
      echo_input=input_text,
      echo_reponse=get_echo_response(input_text))
  return render_template("basic_ui.html")


def get_echo_response(input):
  return f'{CHARACTER_NAME} said {input}'

if __name__ == '__main__':
  app.run(host=SERVICE_HOST, port=SERVER_PORT)

ARG BASE_IMAGE=python:3.10-slim-buster
FROM $BASE_IMAGE
COPY echo_service.py ./
COPY templates/ ./templates/
RUN pip install --upgrade pip && \\
pip install flask
CMD ["python", "echo_service.py"]

<!DOCTYPE html>
<html lang="en">
  <head>
    <title>Welcome to echo service!</title>
  </head>
  <body>
    <h1>Welcome to echo service!</h1>
    <form action="{{ url_for("ui") }}" method="post">
      <label for="input">Input:<label><br>
      <input type="text" id="input" name="input"><br>
    </form>
    <h2>Input:</h2>
    {{ echo_input }}
    <h2>Output:</h2>
    {{ echo_reponse }}
  </body>
</html>

spec:
  containers:
  - name: echo
    image: /tutorial_db/data_schema/tutorial_repository/my_echo_service_image:latest
    env:
      SERVER_PORT: 8000
      CHARACTER_NAME: Bob
    readinessProbe:
      port: 8000
      path: /healthcheck
  endpoints:
  - name: echoendpoint
    port: 8000
    public: true

Como entender a função de serviço¶

Uma função de serviço é um dos métodos de comunicação com seu serviço (consulte Como usar um serviço). Uma função de serviço é uma função definida pelo usuário (UDF) que você associa a um ponto de extremidade de serviço. Quando a função de serviço é executada, ela envia uma solicitação ao ponto de extremidade de serviço associado e recebe uma resposta.

Você cria a seguinte função de serviço executando o comando CREATE FUNCTION com os seguintes parâmetros:

CREATE FUNCTION my_echo_udf (InputText VARCHAR)
  RETURNS VARCHAR
  SERVICE=echo_service
  ENDPOINT=echoendpoint
  AS '/echo';

Observe o seguinte:

• A função my_echo_udf usa uma cadeia de caracteres como entrada e retorna uma cadeia de caracteres.

• A propriedade SERVICE identifica o serviço (echo_service) e a propriedade ENDPOINT identifica o nome do ponto de extremidade amigável (echoendpoint).

• O AS “/echo” especifica o caminho para o serviço. Em echo_service.py, o decorador @app.post associa este caminho à função echo.

Esta função se conecta ao ENDPOINT específico do SERVICE especificado. Quando você invoca esta função, o Snowflake envia uma solicitação para o caminho /echo dentro do contêiner de serviço.

Criação e teste de uma imagem localmente¶

Você pode testar a imagem do Docker localmente antes de carregá-la em um repositório em sua conta Snowflake. Nos testes locais, seu contêiner é executado de forma independente (não é um serviço executado pelo Snowflake).

Para testar a imagem do Docker do Tutorial 1:

1. Para criar uma imagem do Docker, no Docker CLI, execute o seguinte comando:

   docker build --rm -t my_service:local .
   

   Copy

2. Para iniciar seu código, execute o seguinte comando:

   docker run --rm -p 8080:8080 my_service:local
   

   Copy

3. Envie uma solicitação de eco ao serviço usando um dos seguintes métodos:

   • Usando o comando cURL:

     Em outra janela do terminal, usando cURL, envie a seguinte solicitação POST para a porta 8080:

     curl -X POST http://localhost:8080/echo \
       -H "Content-Type: application/json" \
       -d '{"data":[[0, "Hello friend"], [1, "Hello World"]]}'
     

     Copy

     Observe que o corpo da solicitação inclui duas cadeias de caracteres. Este comando cURL envia uma solicitação POST para a porta 8080 na qual o serviço está escutando. O 0 nos dados é o índice da cadeia de caracteres de entrada na lista. O serviço Echo ecoa as cadeias de caracteres de entrada na resposta, conforme mostrado:

     {"data":[[0,"I said Hello Friend"],[1,"I said Hello World"]]}
     

   • Usando um navegador da web:

     1. No seu navegador, no mesmo computador, abra http://localhost:8080/ui.

        Isso envia uma solicitação GET para a porta 8080, na qual o serviço está escutando. O serviço executa a função ui(), que renderiza um formulário HTML conforme mostrado:

        

     2. Insira a cadeia de caracteres “Olá” na caixa Entrada e pressione Retornar.