cat 20_create-clf-kafka-sasl_ssl.sh
kafka_namespace=${KAFKA_NAMESPACE:-openshift-logging}
kafka_user_name="admin"
kafka_user_password="admin-secret"oc delete clf instance -n openshift-logging
oc delete secret kafka-fluentd -n openshift-logging
oc create secret generic kafka-fluentd --from-file=ca-bundle.crt=ca/ca_bundle.crt --from-literal=username=${kafka_user_name} --from-literal=password=${kafka_user_password} --from-literal=sasl_over_ssl=true -n openshift-loggingcat <<EOF |oc create -f -
apiVersion: logging.openshift.io/v1
kind: ClusterLogForwarder
metadata:
  name: instance
  namespace: openshift-logging
spec:
  outputs:
    - name: kafka-app
      url: tls://kafka.${kafka_namespace}.svc.cluster.local:9093/clo-topic
      type: kafka
      secret:
        name: kafka-fluentd
  pipelines:
    - name: test-app
      inputRefs:
      - application
      outputRefs:
      - kafka-app
EOF
 

collector-jnrcs                                 1/2     CrashLoopBackOff   5 (32s ago)     3m37s
 
oc logs collector-jnrcs -c collector
Jan 25 08:35:24.227  INFO vector::app: Log level is enabled. level="debug"
Jan 25 08:35:24.227  INFO vector::sources::host_metrics: PROCFS_ROOT is unset. Using default '/proc' for procfs root.
Jan 25 08:35:24.227  INFO vector::sources::host_metrics: SYSFS_ROOT is unset. Using default '/sys' for sysfs root.
Jan 25 08:35:24.228  INFO vector::app: Loading configs. path=[("/etc/vector/vector.toml", Some(Toml))]
Jan 25 08:35:24.232  INFO vector::sources::kubernetes_logs: Obtained Kubernetes Node name to collect logs for (self). self_node_name="ikansek-qdfkx-master-0.c.openshift-qe.internal"
Jan 25 08:35:24.246 ERROR vector::topology: Configuration error. error=Sink "kafka_app": creating kafka producer failed: Client creation error: Invalid sasl.kerberos.kinit.cmd value: Property not available: "sasl.kerberos.keytab" 

# Rename log stream to "application"
[transforms.application]
type = "remap"
inputs = ["route_container_logs.app"]
source = """
.log_type = "application"
"""



[transforms.test-app]
type = "remap"
inputs = ["application"]
source = """
.
"""
 
# Kafka config
[sinks.kafka_app]
type = "kafka"
inputs = ["test-app"]
bootstrap_servers = "kafka.openshift-logging.svc.cluster.local:9093"
topic = "clo-topic"
 
[sinks.kafka_app.encoding]
codec = "json"
timestamp_format = "rfc3339"
 
# TLS Config
[sinks.kafka_app.tls]
ca_file = "/var/run/ocp-collector/secrets/kafka-fluentd/ca-bundle.crt"
enabled = true
 
# Sasl Config
[sinks.kafka_app.sasl]
user = "admin"
password = "admin-secret"
enabled = true