Overview

IBM® Automation Flink, which is based on Apache Flink®, provides event-processing capabilities.

Event processing involves listening for events from one or more sources, performing meaningful operations such as aggregation and pattern detection, and then sending the processed events to one or more consumers or data stores.

A Flink cluster is an event processor, and a Flink job is an event processing task.

The iaf-flink-operator-controller-manager deployment provides the FlinkCluster APIs.

• Prerequisites
• Connection details for Elasticsearch
• Storage
• Example Flink Cluster CR

Prerequisites

The IBM Automation Flink service needs the following components:

• Certificates for the Flink REST API connectivity under Secret, which is referenced later in the FlinkCluster
• Certificates for the Flink intra-TLS communication under Secret, which is referenced later in the FlinkCluster
• Kafka connection details added in a Secret, which is referenced later in the FlinkCluster
• Elastic connection details added in a Secret, which is referenced later in the FlinkCluster
• A ReadWriteMany (RWX) storage class

• Flink certificate CA, which is created by using the ibm-cert-manager-operator. Following is a sample issuer and certificate. You must later specify the certificate CA in the FlinkCluster CR under the mount called ca-certs.

    apiVersion: cert-manager.io/v1
    kind: Issuer
    metadata:
      name: [cartridgerequirement-name]-ep-ss-issuer
      namespace: [namespace]
    spec:
      selfSigned: {}
  

    apiVersion: cert-manager.io/v1
    kind: Certificate
    metadata:
      name: [cartridgerequirement-name]-ep-ss-ca
      namespace: [namespace]
    spec:
      commonName: eventprocessing.automation.ibm.com
      duration: 2160h0m0s
      isCA: true
      issuerRef:
        kind: Issuer
        name: [cartridgerequirement-name]-ep-ss-issuer
      renewBefore: 720h0m0s
      secretName: [cartridgerequirement-name]-ep-ss-cacert-kp
      usages:
        - cert sign
        - signing
  

• A Secret with only ca.crt field.

    kind: Secret
    apiVersion: v1
    metadata:
      name: [cartridgerequirement-name]-ep-ca-certs
      namespace: [namespace]
    data:
      ca.crt: >-
        [copied ca.crt from [cartridgerequirement-name]-ep-ss-cacert-kp]
    type: Opaque
  

• Flink certificates for Flink/JobManager, Flink/internal communication

  • Issuer

    apiVersion: cert-manager.io/v1
    kind: Issuer
    metadata:
    name: [cartridgerequirement-name]-ep-issuer
    namespace: [namespace]
    spec:
    ca:
      secretName: [cartridgerequirement-name]-ep-ss-cacert-kp
    

  • Certificate for Flink/JobManager

    apiVersion: cert-manager.io/v1
    kind: Certificate
    metadata:
    name: [cartridgerequirement-name]-ep-client-cert
    namespace: [namespace]
    spec:
    commonName: acme-smartdecisions--f76f-eve-29ee-ep-jobmanager
    dnsNames:
      - [cartridgerequirement-name]-ep-jobmanager
      - [cartridgerequirement-name]-ep-jobmanager.[namespace]
      - [cartridgerequirement-name]-ep-jobmanager.[namespace].svc
    duration: 2160h0m0s
    issuerRef:
      kind: Issuer
      name: [cartridgerequirement-name]-ep-issuer
    renewBefore: 720h0m0s
    secretName: [cartridgerequirement-name]-ep-client-cert-kp
    usages:
      - key encipherment
      - server auth
      - digital signature
    

  • Certificate for Flink/internal communication

    apiVersion: cert-manager.io/v1
    kind: Certificate
    metadata:
    name: [cartridgerequirement-name]-ep-internal-cert
    namespace: [namespace]
    spec:
    commonName: acme-smartdecisions--f76f-eve-29ee-ep-taskmanager
    dnsNames:
      - '*.[cartridgerequirement-name]-jobmanager'
      - '*.[cartridgerequirement-name]-ep-jobmanager.[namespace]'
      - '*.[cartridgerequirement-name]-ep-jobmanager.[namespace].svc'
      - '*.[cartridgerequirement-name]-ep-taskmanager'
      - '*.[cartridgerequirement-name]-ep-taskmanager.[namespace]'
      - '*.[cartridgerequirement-name]-ep-taskmanager.[namespace].svc'
    duration: 2160h0m0s
    issuerRef:
      kind: Issuer
      name: [cartridgerequirement-name]-ep-issuer
    renewBefore: 720h0m0s
    secretName: [cartridgerequirement-name]-ep-internal-cert-kp
    usages:
      - key encipherment
      - server auth
      - client auth
      - digital signature
    

• Volume and volume mounts

  • Volume mounts

     volumeMounts:
        - mountPath: /var/iaf/internal/certs
          name: internal-certs
          readOnly: true
        - mountPath: /var/iaf/internal/cacerts
          name: ca-certs
          readOnly: true
        - mountPath: /var/iaf/truststore/
          name: truststore-password
          readOnly: true
        - mountPath: /var/iaf/keystore/
          name: keystore-password
          readOnly: true
        - mountPath: /var/iaf/internal/truststore
          name: internal-truststore-password
          readOnly: true
        - mountPath: /var/iaf/internal/keystore
          name: internal-keystore-password
          readOnly: true
    

  • JobManager volumes

      volumes:
        - name: certs-config
          secret:
            secretName: [cartridgerequirement-name]-ep-client-cert-kp
        - name: internal-certs
          secret:
            secretName: [cartridgerequirement-name]-ep-internal-cert-kp
        - name: ca-certs
          secret:
            secretName: [cartridgerequirement-name]-ep-ca-certs
        - name: truststore-password
          secret:
            secretName: [cartridgerequirement-name]-ep-truststore
        - name: keystore-password
          secret:
            secretName: [cartridgerequirement-name]-ep-keystore
        - name: internal-truststore-password
          secret:
            secretName: [cartridgerequirement-name]-ep-internal-truststore
        - name: internal-keystore-password
          secret:
            secretName: [cartridgerequirement-name]-ep-internal-keystore
    

  • TaskManager volumes

      volumes:
        - name: internal-certs
          secret:
            secretName: [cartridgerequirement-name]-ep-internal-cert-kp
        - name: ca-certs
          secret:
            secretName: [cartridgerequirement-name]-ep-ca-certs
        - name: truststore-password
          secret:
            secretName: [cartridgerequirement-name]-ep-truststore
        - name: keystore-password
          secret:
            secretName: [cartridgerequirement-name]-29ee-ep-keystore
        - name: internal-truststore-password
          secret:
            secretName: [cartridgerequirement-name]-ep-internal-truststore
        - name: internal-keystore-password
          secret:
            secretName: [cartridgerequirement-name]-ep-internal-keystore
    

• Flink ServiceAccount to be manually created in the same namespace as the FlinkCluster. You must add your entitlement key in the service account. The service account name is referenced later in the FlinkCluster CR.

    apiVersion: v1
    kind: ServiceAccount
    metadata:
      name: [service-account-name]
      namespace: [namespace]
    imagePullSecrets:
    - name: ibm-entitlement-key
    commonLabels:
      app.kubernetes.io/component: serviceaccount
  

Connection details for Elasticsearch

For easier access to Elastic runtime from within a Flink Job, you can provide the connection details in the FlinkCluster CR. Specify the following values for environment variables, volumes, and volumeMounts, with the connection credentials.

  envVars:
    - name: ELASTIC_AUTH_TYPE
      value: BASIC
    - name: ELASTIC_BASIC_USERNAME_PATH
      value: /var/iaf/auth/elastic/username
    - name: ELASTIC_BASIC_PASSWORD_PATH
      value: /var/iaf/auth/elastic/password
    - name: ELASTIC_TRUSTSTORE_PATH
      value: /opt/flink/truststore.p12
    - name: ELASTIC_CERTIFICATES
      value: /var/iaf/cacerts/elastic/ca.crt
    - name: ELASTIC_TRUSTSTORE_TYPE
      value: PKCS12
    - name: ELASTIC_TRUSTSTORE_PASSWORD_PATH
      value: /var/iaf/truststore/password
    - name: ELASTIC_TLS_VERSION
      value: TLSv1.2
    - name: ELASTIC_URI
      value: 'https://[location of iaf-system-elasticsearch-es Service]:9200' # e.x. https://iaf-system-elasticsearch-es.iaf:9200

    volumeMounts:
      - mountPath: /var/iaf/auth/elastic
        name: elastic-authentication
      - mountPath: /var/iaf/cacerts/elastic
        name: elastic-certificates

    volumes:
      - name: elastic-authentication
        secret:
          secretName: [cartridge-name]-es-auth
      - name: elastic-certificates
        secret:
          secretName: [automationbase-name]-automationbase-ab-ca

Storage

An RWX, shared PersistentVolumeClaim (PVC) for the Flink JobManagers and TaskManagers provides stateful checkpoint and savepoint for Flink jobs. When you have more than 1 Flink replica, the RWX PVC is mandatory.

For a single replica, you can mount the volume as read/write on a single node.

Note: Make sure that you provision an RWX storage class in your cluster.

In the FlinkCluster CR, provide the following PVC as the volume jobs-storage:

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: icp4adeploy-bai-pvc
spec:
  accessModes:
  - ReadWriteMany
  resources:
    requests:
      storage: 20Gi
  storageClassName: ocs-storagecluster-cephfs

Example Flink Cluster CR

A Flink cluster CR with status is shown in the following example:

apiVersion: flink.automation.ibm.com/v1beta1
kind: FlinkCluster
metadata:
  annotations:
    kubectl.kubernetes.io/last-applied-configuration:
  name: rl-flink-test
  namespace: rl-flink-control
  labels:
    app: flink
    app.kubernetes.io/managed-by: iaf-flink-operator
spec:
  version: v1
status:
  basicAuthSecretName: rl-flink-test-admin-user
  components:
    configMap:
      name: rl-flink-test-configmap
      state: Ready
    jobManagerService:
      name: rl-flink-test-jobmanager
      state: Ready
    jobManagerStatefulSet:
      name: rl-flink-test-jobmanager
      state: Ready
    taskManagerService:
      name: rl-flink-test-taskmanager
      state: Ready
    taskManagerStatefulSet:
      name: rl-flink-test-taskmanager
      state: NotReady
  state: Reconciling
  versions:
    available:
      channels:
        - name: v1
        - name: v1.4
        - name: v1.3
        - name: v1.2
        - name: v1.1
        - name: v1.0
      versions:
        - name: 1.4.0
        - name: 1.3.0
        - name: 1.2.1
        - name: 1.2.0
        - name: 1.1.0
        - name: 1.0.0
    reconciled: 1.4.0