Configure Functions runtime

You can use the following methods to run functions.

  • Thread: Invoke functions threads in functions worker.
  • Process: Invoke functions in processes forked by functions worker.
  • Kubernetes: Submit functions as Kubernetes StatefulSets by functions worker.
note

Pulsar supports adding labels to the Kubernetes StatefulSets and services while launching functions, which facilitates selecting the target Kubernetes objects.

The differences of the thread and process modes are:

  • Thread mode: when a function runs in thread mode, it runs on the same Java virtual machine (JVM) with functions worker.
  • Process mode: when a function runs in process mode, it runs on the same machine that functions worker runs.

Configure thread runtime

It is easy to configure Thread runtime. In most cases, you do not need to configure anything. You can customize the thread group name with the following settings:

  2. functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory
  3. functionRuntimeFactoryConfigs:
  4.   threadGroupName: "Your Function Container Group"

Thread runtime is only supported in Java function.

Configure process runtime

When you enable Process runtime, you do not need to configure anything.

  2. functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.process.ProcessRuntimeFactory
  3. functionRuntimeFactoryConfigs:
  4.   # the directory for storing the function logs
  5.   logDirectory:
  6.   # change the jar location only when you put the java instance jar in a different location
  7.   javaInstanceJarLocation:
  8.   # change the python instance location only when you put the python instance jar in a different location
  9.   pythonInstanceLocation:
  10.   # change the extra dependencies location:
  11.   extraFunctionDependenciesDir:

Process runtime is supported in Java, Python, and Go functions.

Configure Kubernetes runtime

When the functions worker generates Kubernetes manifests and apply the manifests, the Kubernetes runtime works. If you have run functions worker on Kubernetes, you can use the serviceAccount associated with the pod that the functions worker is running in. Otherwise, you can configure it to communicate with a Kubernetes cluster.

The manifests, generated by the functions worker, include a StatefulSet, a Service (used to communicate with the pods), and a Secret for auth credentials (when applicable). The StatefulSet manifest (by default) has a single pod, with the number of replicas determined by the “parallelism” of the function. On pod boot, the pod downloads the function payload (via the functions worker REST API). The pod’s container image is configurable, but must have the functions runtime.

The Kubernetes runtime supports secrets, so you can create a Kubernetes secret and expose it as an environment variable in the pod. The Kubernetes runtime is extensible, you can implement classes and customize the way how to generate Kubernetes manifests, how to pass auth data to pods, and how to integrate secrets.

tip

For the rules of translating Pulsar object names into Kubernetes resource labels, see here.

Basic configuration

It is easy to configure Kubernetes runtime. You can just uncomment the settings of kubernetesContainerFactory in the functions_worker.yaml file. The following is an example.

  2. functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactory
  3. functionRuntimeFactoryConfigs:
  4.   # uri to kubernetes cluster, leave it to empty and it will use the kubernetes settings in function worker
  5.   k8Uri:
  6.   # the kubernetes namespace to run the function instances. it is `default`, if this setting is left to be empty
  7.   jobNamespace:
  8.   # The Kubernetes pod name to run the function instances. It is set to
  9.   # `pf-<tenant>-<namespace>-<function_name>-<random_uuid(8)>` if this setting is left to be empty
  10.   jobName: 
  11.   # the docker image to run function instance. by default it is `apachepulsar/pulsar`
  12.   pulsarDockerImageName:
  13.   # the docker image to run function instance according to different configurations provided by users.
  14.   # By default it is `apachepulsar/pulsar`.
  15.   # e.g:
  16.   # functionDockerImages:
  17.   #   JAVA: JAVA_IMAGE_NAME
  18.   #   PYTHON: PYTHON_IMAGE_NAME
  19.   #   GO: GO_IMAGE_NAME
  20.   functionDockerImages:
  21.   # "The image pull policy for image used to run function instance. By default it is `IfNotPresent`
  22.   imagePullPolicy: IfNotPresent
  23.   # the root directory of pulsar home directory in `pulsarDockerImageName`. by default it is `/pulsar`.
  24.   # if you are using your own built image in `pulsarDockerImageName`, you need to set this setting accordingly
  25.   pulsarRootDir:
  26.   # The config admin CLI allows users to customize the configuration of the admin cli tool, such as:
  27.   # `/bin/pulsar-admin and /bin/pulsarctl`. By default it is `/bin/pulsar-admin`. If you want to use `pulsarctl`
  28.   # you need to set this setting accordingly
  29.   configAdminCLI: 
  30.   # this setting only takes effects if `k8Uri` is set to null. if your function worker is running as a k8 pod,
  31.   # setting this to true is let function worker to submit functions to the same k8s cluster as function worker
  32.   # is running. setting this to false if your function worker is not running as a k8 pod.
  33.   submittingInsidePod: false
  34.   # setting the pulsar service url that pulsar function should use to connect to pulsar
  35.   # if it is not set, it will use the pulsar service url configured in worker service
  36.   pulsarServiceUrl:
  37.   # setting the pulsar admin url that pulsar function should use to connect to pulsar
  38.   # if it is not set, it will use the pulsar admin url configured in worker service
  39.   pulsarAdminUrl:
  40.   # The flag indicates to install user code dependencies. (applied to python package)
  41.   installUserCodeDependencies:
  42.   # The repository that pulsar functions use to download python dependencies
  43.   pythonDependencyRepository:
  44.   # The repository that pulsar functions use to download extra python dependencies
  45.   pythonExtraDependencyRepository:
  46.   # the custom labels that function worker uses to select the nodes for pods
  47.   customLabels:
  48.   # The expected metrics collection interval, in seconds
  49.   expectedMetricsCollectionInterval: 30
  50.   # Kubernetes Runtime will periodically checkback on
  51.   # this configMap if defined and if there are any changes
  52.   # to the kubernetes specific stuff, we apply those changes
  53.   changeConfigMap:
  54.   # The namespace for storing change config map
  55.   changeConfigMapNamespace:
  56.   # The ratio cpu request and cpu limit to be set for a function/source/sink.
  57.   # The formula for cpu request is cpuRequest = userRequestCpu / cpuOverCommitRatio
  58.   cpuOverCommitRatio: 1.0
  59.   # The ratio memory request and memory limit to be set for a function/source/sink.
  60.   # The formula for memory request is memoryRequest = userRequestMemory / memoryOverCommitRatio
  61.   memoryOverCommitRatio: 1.0
  62.   # The port inside the function pod which is used by the worker to communicate with the pod
  63.   grpcPort: 9093
  64.   # The port inside the function pod on which prometheus metrics are exposed
  65.   metricsPort: 9094
  66.   # The directory inside the function pod where nar packages will be extracted
  67.   narExtractionDirectory:
  68.   # The classpath where function instance files stored
  69.   functionInstanceClassPath:
  70.   # Upload the builtin sources/sinks to BookKeeper.
  71.   # True by default.
  72.   uploadBuiltinSinksSources: true
  73.   # the directory for dropping extra function dependencies
  74.   # if it is not an absolute path, it is relative to `pulsarRootDir`
  75.   extraFunctionDependenciesDir:
  76.   # Additional memory padding added on top of the memory requested by the function per on a per instance basis
  77.   percentMemoryPadding: 10
  78.   # The duration (in seconds) before the StatefulSet is deleted after a function stops or restarts.
  79.   # Value must be a non-negative integer. 0 indicates the StatefulSet is deleted immediately.
  80.   # Default is 5 seconds.
  81.   gracePeriodSeconds: 5

If you run functions worker embedded in a broker on Kubernetes, you can use the default settings.

Run standalone functions worker on Kubernetes

If you run functions worker standalone (that is, not embedded) on Kubernetes, you need to configure pulsarSerivceUrl to be the URL of the broker and pulsarAdminUrl as the URL to the functions worker.

For example, both Pulsar brokers and Function Workers run in the pulsar K8S namespace. The brokers have a service called brokers and the functions worker has a service called func-worker. The settings are as follows:

  2. pulsarServiceUrl: pulsar://broker.pulsar:6650 // or pulsar+ssl://broker.pulsar:6651 if using TLS
  3. pulsarAdminUrl: http://func-worker.pulsar:8080 // or https://func-worker:8443 if using TLS

Run RBAC in Kubernetes clusters

If you run RBAC in your Kubernetes cluster, make sure that the service account you use for running functions workers (or brokers, if functions workers run along with brokers) have permissions on the following Kubernetes APIs.

  • services
  • configmaps
  • pods
  • apps.statefulsets

The following is sufficient:

  2. apiVersion: rbac.authorization.k8s.io/v1beta1
  3. kind: ClusterRole
  4. metadata:
  5.   name: functions-worker
  6. rules:
  7. - apiGroups: [""]
  8.   resources:
  9.   - services
  10.   - configmaps
  11.   - pods
  12.   verbs:
  13.   - '*'
  14. - apiGroups:
  15.   - apps
  16.   resources:
  17.   - statefulsets
  18.   verbs:
  19.   - '*'
  20. ---
  21. apiVersion: v1
  22. kind: ServiceAccount
  23. metadata:
  24.   name: functions-worker
  25. ---
  26. apiVersion: rbac.authorization.k8s.io/v1beta1
  27. kind: ClusterRoleBinding
  28. metadata:
  29.   name: functions-worker
  30. roleRef:
  31.   apiGroup: rbac.authorization.k8s.io
  32.   kind: ClusterRole
  33.   name: functions-worker
  34. subjectsKubernetesSec:
  35. - kind: ServiceAccount
  36.   name: functions-worker

If the service-account is not properly configured, an error message similar to this is displayed:

  2. 22:04:27.696 [Timer-0] ERROR org.apache.pulsar.functions.runtime.KubernetesRuntimeFactory - Error while trying to fetch configmap example-pulsar-4qvmb5gur3c6fc9dih0x1xn8b-function-worker-config at namespace pulsar
  3. io.kubernetes.client.ApiException: Forbidden
  4.     at io.kubernetes.client.ApiClient.handleResponse(ApiClient.java:882) ~[io.kubernetes-client-java-2.0.0.jar:?]
  5.     at io.kubernetes.client.ApiClient.execute(ApiClient.java:798) ~[io.kubernetes-client-java-2.0.0.jar:?]
  6.     at io.kubernetes.client.apis.CoreV1Api.readNamespacedConfigMapWithHttpInfo(CoreV1Api.java:23673) ~[io.kubernetes-client-java-api-2.0.0.jar:?]
  7.     at io.kubernetes.client.apis.CoreV1Api.readNamespacedConfigMap(CoreV1Api.java:23655) ~[io.kubernetes-client-java-api-2.0.0.jar:?]
  8.     at org.apache.pulsar.functions.runtime.KubernetesRuntimeFactory.fetchConfigMap(KubernetesRuntimeFactory.java:284) [org.apache.pulsar-pulsar-functions-runtime-2.4.0-42c3bf949.jar:2.4.0-42c3bf949]
  9.     at org.apache.pulsar.functions.runtime.KubernetesRuntimeFactory$1.run(KubernetesRuntimeFactory.java:275) [org.apache.pulsar-pulsar-functions-runtime-2.4.0-42c3bf949.jar:2.4.0-42c3bf949]
  10.     at java.util.TimerThread.mainLoop(Timer.java:555) [?:1.8.0_212]
  11.     at java.util.TimerThread.run(Timer.java:505) [?:1.8.0_212]

Integrate Kubernetes secrets

In order to safely distribute secrets, Pulasr Functions can reference Kubernetes secrets. To enable this, set the secretsProviderConfiguratorClassName to org.apache.pulsar.functions.secretsproviderconfigurator.KubernetesSecretsProviderConfigurator.

You can create a secret in the namespace where your functions are deployed. For example, you deploy functions to the pulsar-func Kubernetes namespace, and you have a secret named database-creds with a field name password, which you want to mount in the pod as an environment variable called DATABASE_PASSWORD. The following functions configuration enables you to reference that secret and mount the value as an environment variable in the pod.

  2. tenant: "mytenant"
  3. namespace: "mynamespace"
  4. name: "myfunction"
  5. topicName: "persistent://mytenant/mynamespace/myfuncinput"
  6. className: "com.company.pulsar.myfunction"
  8. secrets:
  9.   # the secret will be mounted from the `password` field in the `database-creds` secret as an env var called `DATABASE_PASSWORD`
  10.   DATABASE_PASSWORD:
  11.     path: "database-creds"
  12.     key: "password"

Enable token authentication

When you enable authentication for your Pulsar cluster, you need a mechanism for the pod running your function to authenticate with the broker.

The org.apache.pulsar.functions.auth.KubernetesFunctionAuthProvider interface provides support for any authentication mechanism. The functionAuthProviderClassName in function-worker.yml is used to specify your path to this implementation.

Pulsar includes an implementation of this interface for token authentication, and distributes the certificate authority via the same implementation. The configuration is similar as follows:

  2. functionAuthProviderClassName: org.apache.pulsar.functions.auth.KubernetesSecretsTokenAuthProvider

For token authentication, the functions worker captures the token that is used to deploy (or update) the function. The token is saved as a secret and mounted into the pod.

For custom authentication or TLS, you need to implement this interface or use an alternative mechanism to provide authentication. If you use token authentication and TLS encryption to secure the communication with the cluster, Pulsar passes your certificate authority (CA) to the client, so the client obtains what it needs to authenticate the cluster, and trusts the cluster with your signed certificate.

note

If you use tokens that expire when deploying functions, these tokens will expire.

Run clusters with authentication

When you run a functions worker in a standalone process (that is, not embedded in the broker) in a cluster with authentication, you must configure your functions worker to interact with the broker and authenticate incoming requests. So you need to configure properties that the broker requires for authentication or authorization.

For example, if you use token authentication, you need to configure the following properties in the function-worker.yml file.

  2. clientAuthenticationPlugin: org.apache.pulsar.client.impl.auth.AuthenticationToken
  3. clientAuthenticationParameters: file:///etc/pulsar/token/admin-token.txt
  4. configurationMetadataStoreUrl: zk:zookeeper-cluster:2181 # auth requires a connection to zookeeper
  5. authenticationProviders:
  6.  - "org.apache.pulsar.broker.authentication.AuthenticationProviderToken"
  7. authorizationEnabled: true
  8. authenticationEnabled: true
  9. superUserRoles:
  10.   - superuser
  11.   - proxy
  12. properties:
  13.   tokenSecretKey: file:///etc/pulsar/jwt/secret # if using a secret token, key file must be DER-encoded
  14.   tokenPublicKey: file:///etc/pulsar/jwt/public.key # if using public/private key tokens, key file must be DER-encoded
note

You must configure both the Function Worker authorization or authentication for the server to authenticate requests and configure the client to be authenticated to communicate with the broker.

Customize Kubernetes runtime

The Kubernetes integration enables you to implement a class and customize how to generate manifests. You can configure it by setting runtimeCustomizerClassName in the functions-worker.yml file and use the fully qualified class name. You must implement the org.apache.pulsar.functions.runtime.kubernetes.KubernetesManifestCustomizer interface.

The functions (and sinks/sources) API provides a flag, customRuntimeOptions, which is passed to this interface.

To initialize the KubernetesManifestCustomizer, you can provide runtimeCustomizerConfig in the functions-worker.yml file. runtimeCustomizerConfig is passed to the public void initialize(Map<String, Object> config) function of the interface. runtimeCustomizerConfigis different from the customRuntimeOptions as runtimeCustomizerConfig is the same across all functions. If you provide both runtimeCustomizerConfig and customRuntimeOptions, you need to decide how to manage these two configurations in your implementation of KubernetesManifestCustomizer.

Pulsar includes a built-in implementation. To use the basic implementation, set runtimeCustomizerClassName to org.apache.pulsar.functions.runtime.kubernetes.BasicKubernetesManifestCustomizer. The built-in implementation initialized with runtimeCustomizerConfig enables you to pass a JSON document as customRuntimeOptions with certain properties to augment, which decides how the manifests are generated. If both runtimeCustomizerConfig and customRuntimeOptions are provided, BasicKubernetesManifestCustomizer uses customRuntimeOptions to override the configuration if there are conflicts in these two configurations.

Below is an example of customRuntimeOptions.

  2. {
  3.   "jobName": "jobname", // the k8s pod name to run this function instance
  4.   "jobNamespace": "namespace", // the k8s namespace to run this function in
  5.   "extractLabels": {           // extra labels to attach to the statefulSet, service, and pods
  6.     "extraLabel": "value"
  7.   },
  8.   "extraAnnotations": {        // extra annotations to attach to the statefulSet, service, and pods
  9.     "extraAnnotation": "value"
  10.   },
  11.   "nodeSelectorLabels": {      // node selector labels to add on to the pod spec
  12.     "customLabel": "value"
  13.   },
  14.   "tolerations": [             // tolerations to add to the pod spec
  15.     {
  16.       "key": "custom-key",
  17.       "value": "value",
  18.       "effect": "NoSchedule"
  19.     }
  20.   ],
  21.   "resourceRequirements": {  // values for cpu and memory should be defined as described here: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container
  22.     "requests": {
  23.       "cpu": 1,
  24.       "memory": "4G"
  25.     },
  26.     "limits": {
  27.       "cpu": 2,
  28.       "memory": "8G"
  29.     }
  30.   }
  31. }

Run clusters with geo-replication

If you run multiple clusters tied together with geo-replication, it is important to use a different function namespace for each cluster. Otherwise, the function shares a namespace and potentially schedule across clusters.

For example, if you have two clusters: east-1 and west-1, you can configure the functions workers for east-1 and west-1 perspectively as follows.

  2. pulsarFunctionsCluster: east-1
  3. pulsarFunctionsNamespace: public/functions-east-1
  2. pulsarFunctionsCluster: west-1
  3. pulsarFunctionsNamespace: public/functions-west-1

This ensures the two different Functions Workers use distinct sets of topics for their internal coordination.

Configure standalone functions worker

When configuring a standalone functions worker, you need to configure properties that the broker requires, especially if you use TLS. And then Functions Worker can communicate with the broker.

You need to configure the following required properties.

  2. workerPort: 8080
  3. workerPortTls: 8443 # when using TLS
  4. tlsCertificateFilePath: /etc/pulsar/tls/tls.crt # when using TLS
  5. tlsKeyFilePath: /etc/pulsar/tls/tls.key # when using TLS
  6. tlsTrustCertsFilePath: /etc/pulsar/tls/ca.crt # when using TLS
  7. pulsarServiceUrl: pulsar://broker.pulsar:6650/ # or pulsar+ssl://pulsar-prod-broker.pulsar:6651/ when using TLS
  8. pulsarWebServiceUrl: http://broker.pulsar:8080/ # or https://pulsar-prod-broker.pulsar:8443/ when using TLS
  9. useTls: true # when using TLS, critical!