Run Function Mesh

Function Mesh is a collection of functions collaborated together to process data with clearly-defined stages on Kubernetes environments.

This document describes how to run Function Mesh through a .yaml file and submit Pulsar connectors and Pulsar functions through the Function Mesh CRD.

Prerequisites

• Create and connect to a Kubernetes cluster.
• Create a Pulsar cluster in the Kubernetes cluster.
• Install the Function Mesh Operator and CRD into Kubernetes cluster.
• package Pulsar Functions, see package Pulsar Functions.
• Package Pulsar connectors. For details, see package Pulsar connectors.

Steps

To create a Function Mesh, follow these steps.

1. Define a Function Mesh named functionmesh-sample by using a YAML file and save the YAML file functionmesh-sample.yaml.

   apiVersion: compute.functionmesh.io/v1alpha1
   kind: FunctionMesh
   metadata:
     name: functionmesh-sample
   spec:
     sinks:
       - name: sink-sample
         image: streamnative/pulsar-io-elastic-search:2.7.1 # using connector image here
         className: org.apache.pulsar.io.elasticsearch.ElasticSearchSink
         replicas: 1
         maxReplicas: 1
         input:
           topics:
           - persistent://public/default/input
           typeClassName: "[B"
         sinkConfig:
           elasticSearchUrl: "http://quickstart-es-http.default.svc.cluster.local:9200"
           indexName: "my_index"
           typeName: "doc"
           username: "elastic"
           password: "wJ757TmoXEd941kXm07Z2GW3"
         pulsar:
           pulsarConfig: "mesh-test-pulsar"
         resources:
           limits:
             cpu: "0.2"
             memory: 1.1G
           requests:
             cpu: "0.1"
             memory: 1G
         java:
           extraDependenciesDir: random-dir/
           jar: connectors/pulsar-io-elastic-search-2.7.1.nar # the NAR location in image
           jarLocation: "" # leave empty since we will not download package from Pulsar Packages
         clusterName: test-pulsar
         autoAck: true
     functions:
       - name: ex1
         className: exclamation_function.ExclamationFunction
         replicas: 1
         maxReplicas: 1
         logTopic: persistent://public/default/logging-function-log
         input:
           topics:
             - persistent://public/default/functionmesh-input-topic
           typeClassName: java.lang.String
         output:
           topic: persistent://public/default/mid-topic
           typeClassName: java.lang.String
         pulsar:
           pulsarConfig: "mesh-test-pulsar"
         java:
           extraDependenciesDir: random-dir/
           jar: pulsar-functions-api-examples.jar
           jarLocation: public/default/nlu-test-functionmesh-ex1
         # following value must be provided if no auto-filling is enabled
         forwardSourceMessageProperty: true
         autoAck: true
         resources:
           requests:
             cpu: "0.1"
             memory: 1G
           limits:
             cpu: "0.2"
             memory: 1.1G
         clusterName: test-pulsar
   ---
   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: mesh-test-pulsar
   data:
     webServiceURL: http://test-pulsar-broker.default.svc.cluster.local:8080
     brokerServiceURL: pulsar://test-pulsar-broker.default.svc.cluster.local:6650

2. Apply the YAML file to create the Function Mesh.

   kubectl apply -f /path/to/functionmesh-sample.yaml

3. Check whether the Function Mesh is created successfully.

   kubectl get all