Cloud 함수 태스크

Cloud 함수 태스크를 사용하면 통합에서 Google Cloud Functions를 구성하고 실행할 수 있습니다. Google Cloud Functions는 경량형 컴퓨팅 솔루션으로, 서버 또는 런타임 환경을 관리할 필요 없이 Cloud 이벤트에 응답하는 것이 유일한 목적인 독립 실행형 함수를 만들기 위해 사용합니다.

자세한 내용은 Google Cloud Functions 문서를 참조하세요.

시작하기 전에

Cloud 함수 태스크를 구성하기 전에 Google Cloud 프로젝트에서 다음 태스크를 수행해야 합니다.

Cloud Functions API를 사용 설정합니다(cloudfunctions.googleapis.com).
Cloud Functions API 사용 설정
다음 IAM 역할을 주 구성원에 할당합니다.
- Cloud Functions 관리자(roles/cloudfunctions.admin)
- Application Integration 편집자(roles/integrations.integrationEditor)
- 서비스 계정 사용자(roles/iam.serviceAccountUser)
주 구성원에게 역할을 부여하는 방법은 액세스 권한 부여, 변경, 취소를 참조하세요.
Cloud Run 함수에 연결하려면 OAuth 2.0 프로필을 만들었거나 통합에 사용자 관리형 서비스 계정을 연결했는지 확인합니다.

통합에 연결된 서비스 계정이 있는 경우 해당 서비스 계정에 Cloud 함수 호출자 IAM 역할을 할당합니다.
서비스 계정에 역할 부여에 대한 자세한 내용은 서비스 계정에 대한 액세스 관리를 참조하세요.
Cloud 함수 태스크에서는 Google OIDC ID 토큰 유형의 인증 프로필만 지원합니다. Cloud 함수 호출자 IAM 역할이 할당된 서비스 계정을 사용하여 Google OIDC ID 토큰 유형의 인증 프로필을 만듭니다. Cloud 함수 태스크에 인증이 필요하지 않으면 태스크 구성 창의 인증 프로필 필드를 비워 둘 수 있습니다.

통합에 OIDC ID 프로필과 사용자 관리형 서비스 계정이 모두 구성된 경우 기본적으로 OIDC ID 프로필이 인증에 사용됩니다. OIDC ID 프로필 또는 사용자 관리형 서비스 계정이 모두 구성되어 있지 않으면 기본 서비스 계정(service-PROJECT_NUMBER@gcp-sa-apigee.iam.gserviceaccount.com)이 Cloud 함수 태스크를 호출하는 데 사용됩니다.

Google Cloud 프로젝트에서 Apigee Integration에 VPC 서비스 제어가 설정되지 않았는지 확인합니다.
경고: Google Cloud 프로젝트에서 Apigee Integration에 VPC 서비스 제어가 설정된 경우 Cloud 함수가 작동하지 않거나 작동을 중지합니다.

Cloud 함수 태스크 구성

통합에서 Cloud 함수 태스크를 구성하려면 다음 단계를 수행합니다.

Apigee UI에서 Apigee 조직을 선택합니다.
개발 > 통합을 클릭합니다.
기존 통합을 선택하거나 통합 만들기를 클릭하여 새 통합을 만듭니다.
새 통합을 만드는 경우:
1. 통합 만들기 대화상자에 이름과 설명을 입력합니다.
2. 지원되는 리전 목록에서 통합에 대한 리전을 선택합니다.
3. 만들기를 클릭합니다.
통합 디자이너 페이지에서 통합이 열립니다.
통합 설계자 탐색 메뉴에서 +태스크/트리거 추가 > 태스크를 클릭하여 사용 가능한 태스크 목록을 봅니다.
Cloud 함수 요소를 클릭하여 통합 편집기에 배치합니다.
디자이너에서 Cloud 함수 요소를 클릭하여 구성 창을 연 다음 Cloud 함수 구성을 클릭합니다.
Cloud 함수 구성 대화상자에서 다음 중 하나를 선택합니다.
- 기존 함수 연결. 통합에 구성된 기존 Cloud 함수를 연결하려면 이 옵션을 선택합니다. Cloud 함수 트리거 URL을 입력합니다.
  중요: 기존 Cloud 함수의 소스 파일이 Cloud 함수 템플릿에 지정된 형식과 같은지 확인하세요.
- 새 함수 만들기. 통합과 연결된 새 Cloud 함수를 만들려면 이 옵션을 선택합니다. Cloud 함수 이름을 입력하고 드롭다운 목록에서 함수 리전을 선택합니다.
저장을 클릭합니다.
기본 Google Cloud 함수는 Google Cloud 프로젝트에 생성되며 통합과 연결됩니다. 태스크 구성 창에는 Cloud 함수의 트리거 URL 및 태스크 매개변수가 표시됩니다.

Cloud 함수 템플릿

기존 Cloud 함수를 사용하여 Cloud 함수를 구성할 때는 함수의 main.py, task.py, requirements.txt 소스 파일이 다음과 같은 형식이어야 합니다.

task.py

  # Sample Code:
  # print(event.get('task_string_key'))
  # event.set('task_int_array_key', [456, 789]);
  # event.log('some logging')
  
  def run(event):
    """Actual cloud function custom logic.
    Args:
      event : event object in main.py that contains all parameters.
    """
    return

main.py

  """Un-editable platform wrapper which invokes user code."""
import traceback

from flask import json
from flask import jsonify
from task import run

VALUE_NAME = [
    'stringValue', 'intValue', 'doubleValue', 'booleanValue', 'protoValue'
]
ARRAY_VALUE_NAME = {
    'stringArray': 'stringValues',
    'intArray': 'intValues',
    'doubleArray': 'doubleValues',
    'booleanArray': 'booleanValues',
    'protoArray': 'protoValues'
}
VALUE_TYPE_URL = 'type.googleapis.com/google.protobuf.Value'
CLOUD_FUNCTION_EXCEPTION_KEY = 'CloudFunctionException'
CLOUD_FUNCTION_LOGGING_KEY = 'CloudFunctionLogging'


class _Event(object):
  """Event object."""

  def __init__(self, json_payload):
    self._event_params = json_payload.get('eventParameters', dict())
    self._task_params = json_payload.get('taskParameters', dict())
    self._log = []
    print('Event param is ' + str(self._event_params))
    print('Task param is ' + str(self._task_params))

  def set(self, key, value):
    """Set the event parameters key-value.

    Args:
      key: parameter key.
      value: parameter value.
    """
    new_param = self._create_param(key, value)
    param = self._get_param_by_key(key)
    if param is None:
      if 'parameters' not in self._event_params:
        self._event_params['parameters'] = []
      self._event_params['parameters'].append(new_param)
    else:
      param['value'] = new_param['value']

  def _create_param(self, key, value):
    """Create a new parameter with given key value pair.

    Args:
      key: parameter key.
      value: parameter value.

    Returns:
      parameter.
    """
    new_param = {}
    new_param['key'] = key
    if isinstance(value, str):
      new_param['value'] = {'stringValue': value}
    elif isinstance(value, int):
      new_param['value'] = {'intValue': value}
    elif isinstance(value, float):
      new_param['value'] = {'doubleValue': value}
    elif isinstance(value, bool):
      new_param['value'] = {'booleanValue': value}
    elif isinstance(value, dict):
      if 'type@' in value:
        new_param['value'] = {'protoValue': value}
      else:
        new_param['value'] = {
            'protoValue': {
                '@type': 'type.googleapis.com/google.protobuf.Value',
                'value': value
            }
        }
    elif isinstance(value, list):
      if not value:
        raise RuntimeError('Cannot create a param with empty list')
      if any(not isinstance(val, type(value[0])) for val in value):
        print('Not all elements in the list have the same type')
        new_param['value'] = {
            'protoValue': {
                '@type': 'type.googleapis.com/google.protobuf.Value',
                'value': value
            }
        }
      elif isinstance(value[0], str):
        new_param['value'] = {'stringArray': {'stringValues': value}}
      elif isinstance(value[0], int):
        new_param['value'] = {'intArray': {'intValues': value}}
      elif isinstance(value[0], float):
        new_param['value'] = {'doubleArray': {'doubleValues': value}}
      elif isinstance(value[0], bool):
        new_param['value'] = {'booleanArray': {'booleanValues': value}}
      elif isinstance(value[0], dict):
        if all('@type' in val and val['@type'] == value[0]['@type']
               for val in value):
          new_param['value'] = {'protoArray': {'protoValues': value}}
        else:
          new_param['value'] = {
              'protoValue': {
                  '@type': 'type.googleapis.com/google.protobuf.Value',
                  'value': value
              }
          }
      else:
        raise RuntimeError('The type ' + type(value[0]) +
                           ' in the list is not supported')
    else:
      raise RuntimeError('Value ' + str(value) + ' has the type ' +
                         type(value) + ' that is not supported')
    return new_param

  def get(self, key):
    """Get the event parameter value for specified key.

    Args:
      key: parameter key.

    Returns:
      Parameter value.
    """
    param = self._get_param_by_key(key)
    if param is None:
      raise RuntimeError('Can not find param with key ' + key)
    return self._get_param_value(param)

  def _get_param_by_key(self, key):
    """Get the parameter for specified key.

    Args:
      key: parameter key.

    Returns:
      Parameter.
    """
    param = self._get_param_by_key_from_params(key, self._task_params)
    if param is None:
      return self._get_param_by_key_from_params(key, self._event_params)
    value = self._get_param_value(param)
    if isinstance(value, str) and len(value) > 2 and value.startswith(
        '$') and value.endswith('$'):
      return self._get_param_by_key_from_params(value[1:-1], self._event_params)
    return param

  def _get_param_by_key_from_params(self, key, params):
    """Get the parameter for specified key from event parameters.

    Args:
      key: parameter key.
      params: event parameters.

    Returns:
      Parameter.
    """
    if not isinstance(params, dict) or 'parameters' not in params:
      return None
    for param in params['parameters']:
      if param['key'] == key:
        return param
    return None

  def _get_param_value(self, param):
    """Get the parameter value for specified parameter.

    Args:
      param: parameter.

    Returns:
      Parameter value.
    """
    value = param['value']
    if len(value) != 1:
      raise RuntimeError('param does not have size of 1')
    for value_name in VALUE_NAME:
      if value_name in value:
        if value_name == 'protoValue' and value[value_name][
            '@type'] == VALUE_TYPE_URL:
          return value[value_name]['value']
        return value[value_name]
    for array_value_name in ARRAY_VALUE_NAME:
      if array_value_name in value:
        return value[array_value_name][ARRAY_VALUE_NAME[array_value_name]]
    raise RuntimeError('Cannot get value from param ' + str(param))

  def set_error(self):
    """Set the cloud function error to event parameters in order for user to see on IP."""

    self.set(CLOUD_FUNCTION_EXCEPTION_KEY, traceback.format_exc())

  def log(self, message):
    self._log.append(str(message))

  def get_response(self):
    """Get the response that can be returned to IP.

    Returns:
      The response text or any set of values that can be turned into a
      Response object using
      `make_response
      <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
    """
    if self._log:
      self.set(CLOUD_FUNCTION_LOGGING_KEY, self._log)
    res = {
        'eventParameters': self._event_params,
    }
    return jsonify(**json.loads(json.htmlsafe_dumps(res)))


def execute_function(request):
  """Entry point of the cloud function.

  Args:
    request (flask.Request): HTTP request object.

  Returns:
    The response text or any set of values that can be turned into a
    Response object using
    `make_response
    <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
  """
  try:
    request_json = request.get_json(silent=True)
    event = _Event(request_json)
    run(event)
  except:
    event.set_error()
  return event.get_response()

requirements.txt

# Function dependencies, for example:
# package>=version

Cloud 함수 태스크 수정

Apigee Integration에서 Cloud 함수 태스크를 구성하면 Google Cloud 프로젝트에서 기본 HTTP 트리거 Cloud 함수가 생성됩니다.

Cloud 함수 태스크를 수정하려면 다음 단계를 수행합니다.

태스크 구성 창에서 Cloud 함수 열기를 클릭합니다.
Google Cloud Console의 함수 세부정보 페이지로 리디렉션됩니다.
수정을 클릭합니다.
구성 페이지에서 Cloud 함수의 기본 구성 설정을 수정할 수 있습니다. 자세한 내용은 Cloud Functions 구성을 참조하세요.
다음을 클릭하여 Cloud 함수의 소스 코드 파일을 수정합니다.
기본적으로 Cloud 함수에는 다음과 같은 소스 파일이 포함됩니다.
- main.py: 이 파일에는 통합에서 Cloud 함수를 실행하는 초기화 코드가 포함되어 있습니다.
  주의: main.py 파일을 수정하거나 수정하지 마세요.
- task.py: 이 파일에는 Cloud 함수의 실행 코드가 포함되어 있습니다. run(event) 함수 내에서 스크립트를 작성합니다. 이 함수는 Cloud 함수 태스크가 실행될 때 호출됩니다. main.py 파일의 event 객체에는 모든 태스크 매개변수가 포함됩니다.
  스크립트의 통합 수준에서 정의된 변수를 사용하는 방법에 대한 자세한 내용은 액세스 통합 변수를 참조하세요.
배포를 클릭합니다.

통합 변수에 액세스

Cloud 함수의 통합 변수에 액세스하려면 태스크 매개변수 형식의 변수를 Cloud 함수 태스크에 전달해야 합니다. 태스크 매개변수는 키-값 쌍입니다. 여기서 키는 Cloud 함수 소스 파일에 사용된 참조 변수의 이름이고 값은 이 참조 변수가 가리키는 해당 통합 변수 이름입니다. 태스크 구성 창의 태스크 매개변수 섹션에서 태스크를 하나 이상 추가할 수 있습니다.

Cloud 함수에서 통합 변수에 액세스하는 데 사용되는 메서드는 다음과 같습니다.

set: 변수에 값을 씁니다.
get: 변수 값을 읽습니다.

예를 들어 Cloud 함수 소스 파일에 사용할 EmployeeName이라는 통합 변수가 있는 경우 다음 태스크 매개변수를 정의합니다.

Key(키): EmployeeKey
값: EmployeeName

다음 샘플 스크립트는 정의된 통합 변수에 액세스하기 위한 set 및 get 함수의 사용법을 보여줍니다.

def run(event):
  # Read the integration variable EmployeeName using the reference variable EmployeeKey
  value = event.get('EmployeeKey');
  # Change the integration variable EmployeeName value using the reference variable EmployeeKey
  newValue = event.set('EmployeeKey' , 'XYZ');
  # The new value of the integration variable is retained throughout the Cloud Function task.
  return

오류 처리 전략

태스크의 오류 처리 전략은 일시적인 오류로 인해 태스크가 실패할 경우 수행할 태스크를 지정합니다. 오류 처리 전략을 사용하는 방법과 다양한 유형의 오류 처리 전략에 대한 자세한 내용은 오류 처리 전략을 참조하세요.

할당량 및 한도

할당량 및 한도에 대한 자세한 내용은 할당량 및 한도를 참조하세요.

다음 단계

모든 태스크 및 트리거 알아보기
통합을 테스트하고 게시하는 방법 알아보기

오류 처리 알아보기
통합 실행 로그 알아보기