Tutoriel 1 : créer un service Snowpark Container Services¶

Conditions préalables¶

Assurez-vous d’avoir les informations suivantes :

• URL d’entrée du point de terminaison public. Exécutez la commande SHOW ENDPOINTS IN SERVICE pour obtenir l’URL :

  SHOW ENDPOINTS IN SERVICE echo_service;
  

  Copy

• Votre nom de compte Snowflake. Pour plus d’informations, consultez Configuration commune : vérifiez que vous êtes prêt à continuer.

• Votre URL de compte Snowflake : c’est le <nom de compte>.snowflakecomputing.com.

• Nom d’utilisateur dans le compte Snowflake. Il s’agit de l’utilisateur que vous avez choisi dans Configuration courante : créer des objets Snowflake. Vous vous connectez à Snowflake en tant qu’utilisateur et testez l’accès programmatique.

• Nom du rôle : vous avez créé un rôle (test_role) dans le cadre de la configuration commune. L’utilisateur assume ce rôle pour effectuer des actions.

Configuration¶

Suivez les étapes pour communiquer avec le service Echo par programmation. À l’aide du code Python fourni, vous envoyez des requêtes au point de terminaison public exposé par le service Echo.

1. À l’invite de commande, créez un répertoire et accédez-y.

2. Configurez l’authentification par paire de clés pour l’utilisateur.

   1. Générez une paire de clés :

      1. Générez une clé privée. Pour simplifier les étapes de l’exercice, vous générez une clé privée non chiffrée. Vous pouvez également utiliser une clé privée chiffrée, mais vous devrez saisir le mot de passe.

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

         Copy

      2. Générez une clé publique (rsa_key.pub) en référençant la clé privée que vous avez créée.

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

         Copy

   2. Vérifiez que vous disposez de la clé privée et de la clé publique générées dans le répertoire.

   3. Attribuez la clé publique à l’utilisateur que vous utilisez pour tester l’accès programmatique. Cela permet à l’utilisateur de spécifier la clé pour l’authentification.

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

      Copy

3. Enregistrez l’exemple de code fourni dans des fichiers Python.

   1. Enregistrez le code suivant dans 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. Enregistrez le code suivant dans 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

Envoyer des requêtes au point de terminaison de service par programmation¶

Exécutez le code Python access-via-keypair.py pour effectuer l’appel d’entrée vers le point de terminaison public du service 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

Pour plus d’informations sur account-identifier, voir Identificateurs de compte.

Comment fonctionne l’authentification¶

Le code convertit d’abord la paire de clés fournie en un jeton JWT. Il envoie ensuite le jeton JWT à Snowflake pour obtenir un jeton OAuth. Enfin, le code utilise le jeton OAuth pour se connecter à Snowflake et accéder au point de terminaison public. Plus précisément, le code effectue les opérations suivantes :

1. Appelle la fonction _get_token(args) pour générer un jeton JWT de la paire de clés que vous fournissez. L’implémentation de la fonction est illustrée :

   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 est une classe d’aide qui vous est fournie. Notez les points suivants concernant les paramètres que vous fournissez lors de la création de cet objet :

   • Paramètres args.account et args.user : un jeton JWT comporte plusieurs champs (voir format de jeton), iss est l’un des domaines. Cette valeur de champ inclut le nom du compte Snowflake et un nom d’utilisateur. Par conséquent, vous fournissez ces valeurs comme paramètres.

   • Les deux paramètres timedelta fournissent les informations suivantes :

     • lifetime précise le nombre de minutes pendant lesquelles la clé sera valide (60 minutes).

     • renewal_delay spécifie le nombre de minutes à partir de maintenant après lesquelles le générateur JWT doit renouveler le JWT.

2. Appelle la fonction token_exchange() pour se connecter à Snowflake et échanger le jeton JWT contre un jeton 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

   Le code précédent construit un JSON définissant la portée du jeton OAuth, le point de terminaison public accessible à l’aide du rôle spécifié. Ce code fait ensuite une demande POST à Snowflake de transmettre le JSON pour échanger le jeton JWT contre un jeton OAuth (voir Échange de jetons) comme indiqué :

   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. Le code appelle ensuite la fonction connect_to_spcs() permettant de se connecter au point de terminaison public du service Echo. Il fournit l’URL (https://<ingress-URL>/ui) du point de terminaison et du jeton OAuth pour l’authentification.

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

   Copy

   L”url est l”spcs_url que vous avez fournie au programme et le token est le jeton OAuth.

   Le service Echo dans cet exemple sert une page HTML (comme expliqué dans la section précédente). Cet exemple de code imprime simplement le code HTML dans la réponse.

Examen du code du tutoriel 1¶

Le fichier zip que vous avez chargé à l’étape 1 comprend les fichiers suivants :

• Dockerfile

• echo_service.py

• templates/basic_ui.html

Vous utilisez également la spécification de service lors de la création du service. La section suivante explique comment ces composants de code fonctionnent ensemble pour créer le service.

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

Comprendre la fonction de service¶

Une fonction de service est l’une des méthodes utilisées pour communiquer avec votre service (voir Utilisation d’un service). Une fonction de service est une fonction définie par l’utilisateur (UDF) que vous associez à un point de terminaison de service. Lorsque la fonction de service est exécutée, elle envoie une requête au point de terminaison de service associé et reçoit une réponse.

Vous créez la fonction de service suivante en exécutant la commande CREATE FUNCTION avec les paramètres suivants :

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

Remarques :

• La fonction my_echo_udf prend une chaîne en entrée et renvoie une chaîne.

• La propriété SERVICE identifie le service (echo_service) et la propriété ENDPOINT identifie le nom convivial du point de terminaison (echoendpoint).

• Le AS “/echo” spécifie le chemin d’accès au service. Dans echo_service.py, le décorateur @app.post associe ce chemin à la fonction echo.

Cette fonction se connecte à l’élément ENDPOINT spécifique de l’élément SERVICE spécifié. Lorsque vous appelez cette fonction, Snowflake envoie une requête au chemin /echo dans le conteneur de service.

Construire et tester une image localement¶

Vous pouvez tester l’image Docker localement avant de la charger vers un référentiel dans votre compte Snowflake. Dans les tests locaux, votre conteneur fonctionne de manière autonome (il ne s’agit pas d’un service exécuté par Snowflake).

Pour tester l’image Docker du tutoriel 1 :

1. Pour créer une image Docker, dans la CLI Docker, exécutez la commande suivante :

   docker build --rm -t my_service:local .
   

   Copy

2. Pour lancer votre code, exécutez la commande suivante :

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

   Copy

3. Envoyez une requête echo au service en utilisant l’une des méthodes suivantes :

   • Utilisation de la commande cURL :

     Dans une autre fenêtre de terminal, en utilisant cURL, envoyez la requête POST suivante au port 8080 :

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

     Copy

     Notez que le corps de la requête comprend deux chaînes. Cette commande cURL envoie une requête POST au port 8080 sur lequel le service est à l’écoute. Le 0 dans les données est l’indice de la chaîne d’entrée dans la liste. Le service echo répercute les chaînes saisies en réponse, comme indiqué :

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

   • En utilisant un navigateur Web :

     1. Dans votre navigateur, sur le même ordinateur, ouvrez http://localhost:8080/ui.

        Cette opération envoie une requête GET au port 8080, sur lequel le service est à l’écoute. Le service exécute la fonction ui() , qui rend un formulaire HTML comme indiqué :

        

     2. Saisissez la chaîne « Hello » dans la case Entrée et appuyez sur Retour.