14-部署Kafkasource和KafkaChannel

发布时间:2024年01月13日

  • 部署KafkaSource

    • KafkaSource负责将Kafka中的消息记录转为CloudEvents

    • 仅在需要从Kafka中加载消息并提供给Knative Eventing上的应用程序使用时才需要KafkaSource

    • 命令:

      kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-controller.yaml
      
      kubectl apply -f https://mirror.ghproxy.com/https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-source.yaml
      
  • 部署KafkaChannel

    • 负责在Knative Eventing上提供基于Kafka集群的Channel实现,后端基于Kafka Topic

    • https://knative.dev/docs/install/yaml-install/eventing/install-eventing-with-yaml/#verify-the-installation

      kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-controller.yaml
      
      kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-channel.yaml
      
      
    • 如果所使用的Kafka集群是外置的,那需要修改kafka-channel-config的configmap中boostrap server的地址.

      kubectl get cm kafka-channel-config -n knative-eventing -o yaml
      

      在这里插入图片描述

    • 修改Default Channel的default-ch-webhook相关配置: 可以在全局修改,也可以在某个名称空间修改

      apiVersion: v1
      kind: ConfigMap
      metadata:
        name: default-ch-webhook
        namespace: knative-eventing
      data:
        default-ch-config: |
          clusterDefault:
            apiVersion: messaging.knative.dev/v1
            kind: InMemoryChannel
          namespaceDefaults:
            event-demo:
              apiVersion: messaging.knative.dev/v1beta1
              kind: KafkaChannel
              spec:
                numPartitions: 5
                replicationFactor: 1 # Kafka集群的节点数量,决定了其复制因子的可用值上限;但使用的partition数量可按需定义;
      
  • 测试KafkaChannel

    • 创建KafkaChannel

      ~$ kn channel create kc01 --type messaging.knative.dev:v1beta1:KafkaChannel
      
      ~$ kn channel list 
      NAME    TYPE              URL                                                 AGE     READY   REASON 
      kc01    KafkaChannel      http://kc01-kn-channel.default.svc.cluster.local    37s     True 
      

      提示:在没有设置KafkaChannel为默认Channel类型的名称空间下创建KafkaChannel时需要显式指定其类型。

    • 测试KafkaChannel

      • 创建订阅至KafkaChannel/kc01的Subscription

        kn subscription create sub-to-kc01 --channel messaging.knative.dev:v1beta1:KafkaChannel:kc01 --sink ksvc:event-display
        
        说明:此处事先存在一个可用的名为event-display的Knative Service。
        
      • 创建用于模拟Source的Pod

        kubectl run client-$RANDOM --image=ikubernetes/admin-box:v1.2 --rm --restart=Never -it --command -- /bin/bash
        
      • 在Pod的交互式接口发送CloudEvents

        root@client-18448 ~# curl -v "http://kc01-kn-channel.default.svc.cluster.local" -X POST -H "Content-Type: application/cloudevents+json" \
            -d '{"id": "100001", "specversion": "1.0", "type": "com.icloud2native.flow.seq", "source": "Curl", "data": {"message":"Hello from Knative Eventing KafkaChannel"}}'
        
      • 在最后的Sink上查看生成的CloudEvents

        POD=$(kubectl get pods -l serving.knative.dev/configuration=event-display -o jsonpath='{.items[0].metadata.name}')
        
        ~$ kubectl logs $POD
        Defaulted container "user-container" out of: user-container, queue-proxy
        ??  cloudevents.Event
        Context Attributes,
          specversion: 1.0
          type: com.icloud2native.flow.seq
          source: Curl
          id: 100001
        Data,
           {"message":"Hello from Knative Eventing KafkaChannel"}
        
文章来源:https://blog.csdn.net/weixin_38753143/article/details/135569959
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。