struggling with pulsar functions in kubernetes

[jesumyip]

I'm getting an OpenSSLEngine general error in my pulsar function pods. I've narrowed it down to pulsar-admin that's causing this issue. We use cert-manager to secure our Pulsar instance.

Everytime I run pulsar-admin manually to create a function, I understand that this causes Pulsar to create a K8s manifest that deploys a pod to run my function. When I inspect the created pod, I see that it's got a few issues.

1. The shell command in the pod that instructs pulsar-admin to download my function (I'm using the example exclamation function written in Python), is not given any instruction to use a private root CA cert. How do I change this behaviour? I have seen documentation to modify /pulsar/conf/client.conf. But this is a manual step. I would expect Pulsar to actually modify this file for me or somehow use the correct cli parameters in the shell command that calls pulsar-admin. Is there somewhere I can configure this behaviour?
2. I can also see that there is no volume mounted in the pod that has the root CA cert at well. Again, how do I change this behaviour?

Sorry for the basic questions as we are completely new to Pulsar.

[michaeljmarshall]

@jesumyip - in your function worker config, do you have brokerClientTrustCertsFilePath or tlsTrustCertsFilePath set? Based on reading through the code, that configuration is used to read the bytes into the function worker's memory, create a kubernetes secret, then mount that secret into the function pod.

This appears to be documented here: https://pulsar.apache.org/docs/next/functions-worker-run-separately/#enable-tls-encryption.

Here are some of the relevant code references:

pulsar/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java

Lines 607 to 615 in 82237d3

	public String getBrokerClientTrustCertsFilePath() {
	// for compatible, if user do not define brokerClientTrustCertsFilePath, we will use tlsTrustCertsFilePath,
	// otherwise we will use brokerClientTrustCertsFilePath
	if (StringUtils.isNotBlank(brokerClientTrustCertsFilePath)) {
	return brokerClientTrustCertsFilePath;
	} else {
	return tlsTrustCertsFilePath;
	}
	}

pulsar/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java

Lines 774 to 785 in 82237d3

	public byte[] getTlsTrustChainBytes() {
	if (StringUtils.isNotEmpty(getBrokerClientTrustCertsFilePath())
	&& Files.exists(Paths.get(getBrokerClientTrustCertsFilePath()))) {
	try {
	return Files.readAllBytes(Paths.get(getBrokerClientTrustCertsFilePath()));
	} catch (IOException e) {
	throw new IllegalStateException("Failed to read CA bytes", e);
	}
	} else {
	return null;
	}
	}

pulsar/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactory.java

Line 221 in 82237d3

this.serverCaBytes = workerConfig.getTlsTrustChainBytes();

(A couple skipped references, but it's easy to follow in an IDE.)

pulsar/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/KubernetesSecretsTokenAuthProvider.java

Lines 279 to 287 in 82237d3

	@VisibleForTesting
	Map<String, byte[]> buildSecretMap(String token) {
	Map<String, byte[]> valueMap = new HashMap<>();
	valueMap.put(FUNCTION_AUTH_TOKEN, token.getBytes());
	if (caBytes != null) {
	valueMap.put(FUNCTION_CA_CERT, caBytes);
	}
	return valueMap;
	}

[michaeljmarshall]

How are you deploying Pulsar? It looks like you're using the helm chart. Which version, and do you mind sharing some/all of your values file?

[jesumyip]

@michaeljmarshall Yes, using helm. Our ops team created a helm subchart that combined pulsar,redis, and istio.

Here are the values we use:

pulsar:
  enabled: true
  namespace: datastream
  initialize: true
  persistence: true
  volumes:
    persistence: true
    local_storage: false
  components:
    # zookeeper
    zookeeper: true
    # bookkeeper
    bookkeeper: true
    # bookkeeper - autorecovery
    autorecovery: true
    # broker
    broker: true
    # functions
    functions: true
    # proxy
    proxy: true
    # toolset
    toolset: true
    # pulsar manager
    pulsar_manager: false
  monitoring:
    # monitoring - prometheus
    prometheus: true
    # monitoring - grafana
    grafana: true
    # monitoring - node_exporter
    node_exporter: true
    # alerting - alert-manager
    alert_manager: true
  certs:
    internal_issuer:
      # use cert-manager to provision TLS certificates
      enabled: true
  tls:
    enabled: true
    common:
      organization:
        - "Test Demo"
    proxy:
      enabled: true
    broker:
      enabled: true
    bookie:
      enabled: true
    zookeeper:
      enabled: true
    autorecovery:
      enabled: true
    toolset:
      enabled: true
  auth:
    authentication:
      enabled: true
      jwt:
        # set to true if generated keys are symmetric; otherwise, false
        usingSecretKey: false
    authorization:
      enabled: true
    superUsers:
      # broker to broker communication
      broker: "broker-admin"
      # proxy to broker communication
      proxy: "proxy-admin"
      # pulsar-admin client to broker/proxy communication
      client: "admin"
  # component settings
  zookeeper:
    nodeSelector:
      workload: stream
    volumes:
      persistence: true
      data:
        name: data
        size: 20Gi
        local_storage: false
        storageClassName: regional-storage-pd-balanced
  bookkeeper:
    nodeSelector:
      workload: stream
    affinity:
      type: preferredDuringSchedulingIgnoredDuringExecution
    volumes:
      persistence: true
      useSingleCommonVolume:
      journal:
        name: journal
        size: 10Gi
        local_storage: false
        storageClassName: regional-storage-pd-balanced
      ledgers:
        name: ledgers
        size: 50Gi
        local_storage: false
        storageClassName: regional-storage-pd-balanced
  autorecovery:
    nodeSelector:
      workload: stream
    affinity:
      type: preferredDuringSchedulingIgnoredDuringExecution
  broker:
    nodeSelector:
      workload: stream
  proxy:
    nodeSelector:
      workload: stream
    autoscaling:
      enabled: true
      minReplicas: 3
      maxReplicas: 6
    affinity:
      anti_affinity: true
      type: preferredDuringSchedulingIgnoredDuringExecution
    service:
      annotations:
        networking.gke.io/load-balancer-type: "Internal"
      type: LoadBalancer
    ingress:
      enabled: false
  dashboard:
    nodeSelector:
      workload: stream
  toolset:
    nodeSelector:
      workload: stream
  prometheus:
    nodeSelector:
      workload: stream
    volumes:
      persistence: true
      data:
        name: data
        size: 10Gi
        local_storage: false
        storageClassName: regional-storage-pd-balanced
  grafana:
    nodeSelector:
      workload: stream
    service:
      type: ClusterIP
    ingress:
      enabled: false
  pulsar_manager:
    nodeSelector:
      workload: stream
  
redis-ha:
  enabled: true
  nodeSelector:
      workload: cache

local-storage-provisioner:
  enabled: false

gateway:
  service:
    # Type of service. Set to "None" to disable the service entirely
    type: None
    ports:
    - name: status-port
      port: 15021
      protocol: TCP
      targetPort: 15021
    - name: http2
      port: 80
      protocol: TCP
      targetPort: 80
    - name: https
      port: 443
      protocol: TCP
      targetPort: 443
    - name: pulsar
      port: 6650
      protocol: TCP
      targetPort: 6650
    - name: pulsar-ssl
      port: 6651
      protocol: TCP
      targetPort: 6651
    annotations:
      networking.gke.io/load-balancer-type: "Internal"
    loadBalancerIP: "<redacted>"
    loadBalancerSourceRanges: [<redacted>]
    externalTrafficPolicy: ""
    externalIPs: []

[michaeljmarshall]

Thanks, that is helpful. I looked closer at the function worker integration in the helm chart, because you're enabling tls in the broker, the helm chart should be adding tlsTrustCertsFilePath to the broker's config map. I haven't verified, but I would expect that to get put into the function runtime. Can you share the error you're observing in the function?

[jesumyip]

I'm getting an error in the pod that runs the function. The error is

General OpenSslEngine problem

I then modified the pod so that instead of running all the shell commands that is by default in it, I paused the pod and did a kubectl exec into the pod. Then I manually ran the shell commands one by one. I found that pulsar-admin function download was the one throwing this error.

[michaeljmarshall]

An actual stack trace would be very helpful. Can you share what you're observing?

[phijohns-tibco]

What version of Pulsar are you using?

I'm getting the same error with pulsar-admin. I'm not in kubernetes but bare metal.

For me this only effects Pulsar 2.9.3. I had no problem with pulsar-admin with either 2.9.2 or 2.10.2 connecting to the same broker.

[phijohns-tibco]

Just in case this helps you out. The solution for my error was to update my version of Java 8. For me it was an invalid certificate. Notice that I said invalid not bad. The build of Java 8 that I came with the O/S was unable to use the certificate I generated with openssl. The version of OpenSSL used was 1.1.1k.

[phijohns-tibco]

java.util.concurrent.CompletionException: org.apache.pulsar.client.admin.PulsarAdminException: java.util.concurrent.CompletionException: org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector$RetryException: Could not complete the operation. Number of retries has been exhausted. Failed reason: General OpenSslEngine problem
at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:331)
at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:346)
at java.base/java.util.concurrent.CompletableFuture$BiRelay.tryFire(CompletableFuture.java:1423)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
at org.apache.pulsar.client.admin.internal.TopicsImpl$5.failed(TopicsImpl.java:287)
at org.glassfish.jersey.client.JerseyInvocation$1.failed(JerseyInvocation.java:882)
at org.glassfish.jersey.client.ClientRuntime.processFailure(ClientRuntime.java:247)
at org.glassfish.jersey.client.ClientRuntime.processFailure(ClientRuntime.java:242)
at org.glassfish.jersey.client.ClientRuntime.access$100(ClientRuntime.java:62)
at org.glassfish.jersey.client.ClientRuntime$2.lambda$failure$1(ClientRuntime.java:178)
at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)
at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)
at org.glassfish.jersey.internal.Errors.process(Errors.java:292)
at org.glassfish.jersey.internal.Errors.process(Errors.java:274)
at org.glassfish.jersey.internal.Errors.process(Errors.java:244)
at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:288)
at org.glassfish.jersey.client.ClientRuntime$2.failure(ClientRuntime.java:178)
at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$apply$1(AsyncHttpConnector.java:218)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$retryOperation$4(AsyncHttpConnector.java:277)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
at org.asynchttpclient.netty.NettyResponseFuture.abort(NettyResponseFuture.java:273)
at org.asynchttpclient.netty.channel.NettyConnectListener.onFailure(NettyConnectListener.java:181)
at org.asynchttpclient.netty.channel.NettyConnectListener$1.onFailure(NettyConnectListener.java:151)
at org.asynchttpclient.netty.SimpleFutureListener.operationComplete(SimpleFutureListener.java:26)
at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578)
at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:571)
at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:550)
at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491)
at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:616)
at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:609)
at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:117)
at io.netty.handler.ssl.SslHandler.handleUnwrapThrowable(SslHandler.java:1258)
at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1240)
at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1284)
at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:510)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:449)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:279)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:995)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:829)