자습서 1: Snowpark Container Services 서비스 생성¶

전제 조건¶

다음 정보가 있는지 확인합니다.

• 공용 엔드포인트의 수신 URL. SHOW ENDPOINTS IN SERVICE 명령을 실행하여 URL을 가져옵니다.

  SHOW ENDPOINTS IN SERVICE echo_service;
  

  Copy

• 사용자의 Snowflake 계정 이름. 자세한 내용은 공통 설정: 계속할 준비가 되었는지 확인합니다 섹션을 참조하십시오.

• 사용자의 Snowflake 계정 URL: <acctname>.snowflakecomputing.com 입니다.

• Snowflake 계정의 사용자 이름. 공통 설정: Snowflake 오브젝트 만들기 섹션에서 선택한 사용자입니다. 이 사용자로 Snowflake에 로그인하여 프로그래밍 방식의 액세스를 테스트합니다.

• 역할 이름: 공통 설정의 일부로 역할(test_role)을 생성했습니다. 사용자는 이 역할을 맡아 작업을 수행합니다.

설정¶

다음 단계를 따라 Echo 서비스와 프로그래밍적으로 통신합니다. 제공된 Python 코드를 사용하여 Echo 서비스가 노출하는 공개 엔드포인트에 요청을 보냅니다.

1. 명령 프롬프트에서 디렉터리를 생성하고 해당 디렉터리로 이동합니다.

2. 사용자에 대한 키 페어 인증을 구성합니다.

   1. 키 페어 를 생성합니다.

      1. 개인 키를 생성합니다. 연습 단계를 단순화하려면 암호화되지 않은 개인 키를 생성합니다. 암호화된 개인 키를 사용할 수도 있지만 비밀번호를 입력해야 합니다.

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

         Copy

      2. 생성한 개인 키를 참조하여 공개 키(rsa_key.pub)를 생성합니다.

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

         Copy

   2. 디렉터리에 개인 키와 공개 키가 생성되었는지 확인합니다.

   3. 프로그래밍 방식 액세스를 테스트하는 데 사용하는 사용자에게 공개 키를 할당합니다. 이를 통해 사용자는 인증 키를 지정할 수 있습니다.

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

      Copy

3. 제공된 샘플 코드를 Python 파일로 저장합니다.

   1. 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. 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

서비스 엔드포인트에 프로그래밍 방식으로 요청 보내기¶

access-via-keypair.py Python 코드를 실행하여 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

account-identifier 에 대한 자세한 내용은 계정 식별자 를 참조하십시오.

인증 작동 방식¶

코드는 우선 공급자가 제공한 키 페어를 JWT 토큰으로 변환합니다. 그런 다음 JWT 토큰을 Snowflake로 전송하여 OAuth 토큰을 얻습니다. 마지막으로 코드는 OAuth 토큰을 사용하여 Snowflake에 연결하고 공용 엔드포인트에 액세스합니다. 구체적으로 코드는 다음을 수행합니다.

1. _get_token(args) 함수를 호출하여 공급자가 제공한 키 페어에서 JWT 토큰을 생성합니다. 함수 구현은 다음과 같습니다.

   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 는 사용자에게 제공되는 도우미 클래스입니다. 이 오브젝트를 생성할 때 제공하는 매개 변수에 대해 다음 사항을 참고합니다.

   • args.account 및 args.user 매개 변수: JWT 토큰에는 여러 개의 필드가 있으며(토큰 형식 참조), iss 는 그 중 하나입니다. 이 필드 값에는 Snowflake 계정 이름과 사용자 이름이 포함됩니다. 따라서 이러한 값을 매개 변수로 제공합니다.

   • 두 개의 timedelta 매개 변수는 다음 정보를 제공합니다.

     • lifetime 은 키가 유효한 시간(60분)을 지정합니다.

     • renewal_delay 는 JWT 생성기가 JWT를 갱신해야 하는 시간(분)을 지정합니다.

2. token_exchange() 함수를 호출하여 Snowflake에 연결하고 OAuth 토큰을 JWT 토큰으로 교환합니다.

   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

   앞의 코드는 지정된 역할을 사용하여 액세스할 수 있는 공개 엔드포인트인 OAuth 토큰의 범위를 설정하는 JSON을 구성합니다. 그런 다음 이 코드는 그림과 같이 JSON을 통과하는 Snowflake에 POST를 요청하여 JWT 토큰을 OAuth 토큰으로 교환합니다(토큰 교환 참조).

   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. 그런 다음 코드는 connect_to_spcs() 함수를 호출하여 Echo 서비스의 공용 엔드포인트에 연결합니다. 엔드포인트의 URL(https://<수신-URL>/ui)와 인증을 위한 OAuth 토큰을 제공합니다.

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

   Copy

   url 은 사용자가 프로그램에 제공한 spcs_url 이고 token 은 OAuth 토큰입니다.

   이 예제의 Echo 서비스는 HTML 페이지를 제공합니다(이전 섹션에서 설명한 대로). 이 샘플 코드는 단순히 응답에 HTML을 출력합니다.

자습서 1 코드 검토¶

1단계에서 다운로드한 zip 파일에는 다음 파일이 포함됩니다.

• Dockerfile

• echo_service.py

• templates/basic_ui.html

서비스를 만들 때 서비스 사양도 사용합니다. 다음 섹션에서는 이러한 코드 구성 요소가 함께 작동하여 서비스를 만드는 방법을 설명합니다.

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

서비스 함수 이해하기¶

서비스 함수는 서비스와 통신하는 방법 중 하나입니다(서비스 사용하기 참조). 서비스 함수는 서비스 엔드포인트와 연결하는 사용자 정의 함수(UDF)입니다. 서비스 함수가 실행되면 연결된 서비스 엔드포인트에 요청을 보내고 응답을 받습니다.

다음 매개 변수와 함께 CREATE FUNCTION 명령을 실행하여 다음 서비스 함수를 생성합니다.

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

다음 사항을 참고하십시오.

• my_echo_udf 함수는 문자열을 입력값으로 받아 문자열을 반환합니다.

• SERVICE 속성은 서비스(echo_service)를 식별하고, ENDPOINT 속성은 사용자에게 친숙한 엔드포인트 이름(echoendpoint)을 식별합니다.

• AS ‘/echo’는 서비스 경로를 지정합니다. echo_service.py 에서 @app.post 데코레이터는 이 경로를 echo 함수와 연결합니다.

이 함수는 지정된 SERVICE의 특정 ENDPOINT와 연결합니다. 이 함수를 호출하면 Snowflake가 서비스 컨테이너 내부의 /echo 경로로 요청을 보냅니다.

로컬에서 이미지 만들기 및 테스트하기¶

Docker 이미지를 Snowflake 계정의 리포지토리에 업로드하기 전에 로컬에서 테스트할 수 있습니다. 로컬 테스트에서는 컨테이너가 독립형으로 실행됩니다(Snowflake가 실행하는 서비스가 아님).

자습서 1 Docker 이미지를 테스트하는 방법은 다음과 같습니다.

1. Docker 이미지를 생성하려면 Docker CLI에서 다음 명령을 실행하십시오.

   docker build --rm -t my_service:local .
   

   Copy

2. 코드를 실행하려면 다음 명령을 실행하십시오.

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

   Copy

3. 다음 방법 중 하나를 사용하여 서비스에 에코 요청을 보냅니다.

   • cURL 명령 사용:

     다른 터미널 창에서 cURL을 사용하여 다음 POST 요청을 포트 8080에 보냅니다.

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

     Copy

     요청 본문에는 다음 두 문자열이 포함됩니다. 이 cURL 명령은 서비스가 수신 대기 중인 포트 8080에 POST 요청을 보냅니다. 데이터에 있는 0은 목록에 있는 입력 문자열의 인덱스입니다. Echo 서비스는 다음과 같이 입력 문자열을 응답으로 표시합니다.

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

   • 웹 브라우저 사용:

     1. 브라우저에서는 동일한 컴퓨터에서 http://localhost:8080/ui 를 엽니다.

        그러면 서비스가 수신 대기 중인 포트 8080으로 GET 요청이 전송됩니다. 서비스는 다음과 같이 HTML 양식을 렌더링하는 ui() 함수를 실행합니다.

        

     2. Input 상자에 문자열 “Hello”를 입력하고 Return 을 누릅니다.