使用 Camel K 开发和测试集成

流程

1. 确保 Camel K Operator 在 OpenShift 集群上运行，例如：

   oc get pod

   NAME                               READY   STATUS    RESTARTS   AGE
   camel-k-operator-86b8d94b4-pk7d6   1/1     Running   0          6m28s

2. 输入 kamel run 命令，以在 OpenShift 上的云中运行集成。例如：

   kamel run hello.camelk.yaml

   integration "hello" created

3. 输入 kamel get 命令来检查集成的状态：

   kamel get

   NAME   PHASE        KIT
   hello  Building Kit kit-bqatqib5t4kse5vukt40

4. 输入 kamel describe 命令，以打印有关集成的详细信息：

   kamel describe integration hello

   Name:                hello
   Namespace:           myproject
   Creation Timestamp:  Fri, 13 Aug 2021 16:23:21 +0200
   Phase:               Building Kit
   Runtime Version:     1.7.1.fuse-800025-redhat-00001
   Kit:                 myproject/kit-c4ci6mbe9hl5ph5c9sjg
   Image:
   Version:             1.6.6
   Dependencies:
     camel:core
     camel:log
     camel:timer
     mvn:org.apache.camel.k:camel-k-runtime
     mvn:org.apache.camel.quarkus:camel-quarkus-yaml-dsl
   Sources:
     Name                        Language  Compression  Ref  Ref Key
     camel-k-embedded-flow.yaml  yaml      false
   Conditions:
     Type                          Status  Reason                        Message
     IntegrationPlatformAvailable  True    IntegrationPlatformAvailable  myproject/camel-k
     IntegrationKitAvailable       True    IntegrationKitAvailable       kit-c4ci6mbe9hl5ph5c9sjg
     CronJobAvailable              False   CronJobNotAvailableReason     different controller strategy used (deployment)
     DeploymentAvailable           True    DeploymentAvailable           deployment name is hello
     KnativeServiceAvailable       False   KnativeServiceNotAvailable    different controller strategy used (deployment)
     Ready                         True    ReplicaSetReady

5. 输入 kamel log 命令将日志输出到 stdout ：

   kamel log hello

   ...
   [1] 2021-08-13 14:37:15,860 INFO [info] (Camel (camel-1) thread #0 - timer://yaml) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Hello Camel K from yaml]
   ...

6. 按 Ctrl-C 终止终端中的日志记录。

7. 输入 kamel delete 以删除 OpenShift 上部署的集成：

   kamel delete hello

   Integration hello deleted

流程

1. 输入 kamel run 命令，并使用 --property 选项指定日志级别。例如：

   kamel run --dev --property 'quarkus.log.category."org.apache.camel.support".level'=DEBUG Basic.java
   
   ...
   integration "basic" created
       Progress: integration "basic" in phase Initialization
       Progress: integration "basic" in phase Building Kit
       Progress: integration "basic" in phase Deploying
       Condition "IntegrationPlatformAvailable" is "True" for Integration basic: myproject/camel-k
       Integration basic in phase "Initialization"
       Integration basic in phase "Building Kit"
       Integration basic in phase "Deploying"
       Condition "IntegrationKitAvailable" is "True" for Integration basic: kit-c4dn5l62v9g3aopkocag
       Condition "DeploymentAvailable" is "True" for Integration basic: deployment name is basic
       Condition "CronJobAvailable" is "False" for Integration basic: different controller strategy used (deployment)
       Progress: integration "basic" in phase Running
       Condition "KnativeServiceAvailable" is "False" for Integration basic: different controller strategy used (deployment)
       Integration basic in phase "Running"
       Condition "Ready" is "False" for Integration basic
       Condition "Ready" is "True" for Integration basic
       [1] Monitoring pod basic-575b97f64b-7l5rl
       [1] 2021-08-17 08:35:22,906 DEBUG [org.apa.cam.sup.LRUCacheFactory] (main) Creating DefaultLRUCacheFactory
       [1] 2021-08-17 08:35:23,132 INFO  [org.apa.cam.k.Runtime] (main) Apache Camel K Runtime 1.7.1.fuse-800025-redhat-00001
       [1] 2021-08-17 08:35:23,134 INFO  [org.apa.cam.qua.cor.CamelBootstrapRecorder] (main) bootstrap runtime: org.apache.camel.quarkus.main.CamelMainRuntime
       [1] 2021-08-17 08:35:23,224 INFO  [org.apa.cam.k.lis.SourcesConfigurer] (main) Loading routes from: SourceDefinition{name='Basic', language='java', location='file:/etc/camel/sources/Basic.java', }
       [1] 2021-08-17 08:35:23,232 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found RoutesBuilderLoader: org.apache.camel.dsl.java.joor.JavaRoutesBuilderLoader via: META-INF/services/org/apache/camel/java
       [1] 2021-08-17 08:35:23,232 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using RoutesBuilderLoader: org.apache.camel.dsl.java.joor.JavaRoutesBuilderLoader@68dc098b
       [1] 2021-08-17 08:35:23,236 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found ResourceResolver: org.apache.camel.impl.engine.DefaultResourceResolvers$FileResolver via: META-INF/services/org/apache/camel/file
       [1] 2021-08-17 08:35:23,237 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using ResourceResolver: org.apache.camel.impl.engine.DefaultResourceResolvers$FileResolver@5b67bb7e
       [1] 2021-08-17 08:35:24,320 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Lookup Language with name simple in registry. Found: org.apache.camel.language.simple.SimpleLanguage@74d7184a
       [1] 2021-08-17 08:35:24,328 DEBUG [org.apa.cam.sup.EventHelper] (main) Ignoring notifying event Initializing CamelContext: camel-1. The EventNotifier has not been started yet: org.apache.camel.quarkus.core.CamelManagementEventBridge@3301500b
       [1] 2021-08-17 08:35:24,336 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Lookup Component with name timer in registry. Found: org.apache.camel.component.timer.TimerComponent@3ef41c66
       [1] 2021-08-17 08:35:24,342 DEBUG [org.apa.cam.sup.DefaultComponent] (main) Creating endpoint uri=[timer://java?period=1000], path=[java]
       [1] 2021-08-17 08:35:24,350 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found ProcessorFactory: org.apache.camel.processor.DefaultProcessorFactory via: META-INF/services/org/apache/camel/processor-factory
       [1] 2021-08-17 08:35:24,351 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using ProcessorFactory: org.apache.camel.processor.DefaultProcessorFactory@704b2127
       [1] 2021-08-17 08:35:24,369 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Found InternalProcessorFactory: org.apache.camel.processor.DefaultInternalProcessorFactory via: META-INF/services/org/apache/camel/internal-processor-factory
       [1] 2021-08-17 08:35:24,369 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Detected and using InternalProcessorFactory: org.apache.camel.processor.DefaultInternalProcessorFactory@4f8caaf3
       [1] 2021-08-17 08:35:24,442 DEBUG [org.apa.cam.sup.ResolverHelper] (main) Lookup Component with name log in registry. Found: org.apache.camel.component.log.LogComponent@46b695ec
       [1] 2021-08-17 08:35:24,444 DEBUG [org.apa.cam.sup.DefaultComponent] (main) Creating endpoint uri=[log://info], path=[info]
       [1] 2021-08-17 08:35:24,461 DEBUG [org.apa.cam.sup.EventHelper] (main) Ignoring notifying event Initialized CamelContext: camel-1. The EventNotifier has not been started yet: org.apache.camel.quarkus.core.CamelManagementEventBridge@3301500b
       [1] 2021-08-17 08:35:24,467 DEBUG [org.apa.cam.sup.DefaultProducer] (main) Starting producer: Producer[log://info]
       [1] 2021-08-17 08:35:24,469 DEBUG [org.apa.cam.sup.DefaultConsumer] (main) Build consumer: Consumer[timer://java?period=1000]
       [1] 2021-08-17 08:35:24,475 DEBUG [org.apa.cam.sup.DefaultConsumer] (main) Starting consumer: Consumer[timer://java?period=1000]
       [1] 2021-08-17 08:35:24,481 INFO  [org.apa.cam.imp.eng.AbstractCamelContext] (main) Routes startup summary (total:1 started:1)
       [1] 2021-08-17 08:35:24,481 INFO  [org.apa.cam.imp.eng.AbstractCamelContext] (main)     Started java (timer://java)
       [1] 2021-08-17 08:35:24,482 INFO  [org.apa.cam.imp.eng.AbstractCamelContext] (main) Apache Camel 3.10.0.fuse-800010-redhat-00001 (camel-1) started in 170ms (build:0ms init:150ms start:20ms)
       [1] 2021-08-17 08:35:24,487 INFO  [io.quarkus] (main) camel-k-integration 1.6.6 on JVM (powered by Quarkus 1.11.7.Final-redhat-00009) started in 2.192s.
       [1] 2021-08-17 08:35:24,488 INFO  [io.quarkus] (main) Profile prod activated.
       [1] 2021-08-17 08:35:24,488 INFO  [io.quarkus] (main) Installed features: [camel-bean, camel-core, camel-java-joor-dsl, camel-k-core, camel-k-runtime, camel-log, camel-support-common, camel-timer, cdi]
       [1] 2021-08-17 08:35:25,493 INFO  [info] (Camel (camel-1) thread #0 - timer://java) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Hello Camel K from java]
       [1] 2021-08-17 08:35:26,479 INFO  [info] (Camel (camel-1) thread #0 - timer://java) Exchange[ExchangePattern: InOnly, BodyType: String, Body: Hello Camel K from java]
   ...

2. 按 Ctrl-C 终止终端中的日志记录。

流程

1. 输入以下命令检查 openshift-monitoring 项目中是否存在 cluster-monitoring-config ConfigMap 对象：

   $ oc -n openshift-monitoring get configmap cluster-monitoring-config

2. 如果不存在，请创建 cluster-monitoring-config ConfigMap：

   $ oc -n openshift-monitoring create configmap cluster-monitoring-config

3. 编辑 cluster-monitoring-config ConfigMap：

   $ oc -n openshift-monitoring edit configmap cluster-monitoring-config

4. 在 data:config.yaml: 下，将 enableUserWorkload 设置为 true ：

   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: cluster-monitoring-config
     namespace: openshift-monitoring
   data:
     config.yaml: |
       enableUserWorkload: true

流程

1. 输入以下命令运行启用了 Prometheus 特征的 Camel K 集成：

   kamel run myIntegration.java -t prometheus.enabled=true

   另外，您可以通过更新集成平台来全局启用 Prometheus trait，如下所示：

   $ oc patch ip camel-k --type=merge -p '{"spec":{"traits":{"prometheus":{"configuration":{"enabled":true}}}}}'

2. 查看 Prometheus 中 Camel K 集成指标的监控。例如，对于嵌入式 Prometheus，在 OpenShift 管理员或开发人员 Web 控制台中选择 Monitoring > Metrics。
3. 输入您要查看的 Camel K 指标。例如，在 Administrator 控制台中，在 Insert Metric at Cursor 下，输入 application_camel_context_uptime_seconds，然后单击 Run Queries。
4. 点 Add Query 查看额外的指标。

流程

1. 使用 Camel MicroProfile Metrics 组件注解在 Camel 集成代码中注册自定义指标。以下示例显示了 Metrics.java 集成：

   // camel-k: language=java trait=prometheus.enabled=true dependency=mvn:org.my/app:1.0 1
   
   import org.apache.camel.Exchange;
   import org.apache.camel.LoggingLevel;
   import org.apache.camel.builder.RouteBuilder;
   import org.apache.camel.component.microprofile.metrics.MicroProfileMetricsConstants;
   
   import javax.enterprise.context.ApplicationScoped;
   
   @ApplicationScoped
   public class Metrics extends RouteBuilder {
   
      @Override
      public void configure() {
           onException()
               .handled(true)
               .maximumRedeliveries(2)
               .logStackTrace(false)
               .logExhausted(false)
               .log(LoggingLevel.ERROR, "Failed processing ${body}")
               // Register the 'redelivery' meter
               .to("microprofile-metrics:meter:redelivery?mark=2")
               // Register the 'error' meter
               .to("microprofile-metrics:meter:error"); 2
   
           from("timer:stream?period=1000")
               .routeId("unreliable-service")
               .setBody(header(Exchange.TIMER_COUNTER).prepend("event #"))
               .log("Processing ${body}...")
               // Register the 'generated' meter
               .to("microprofile-metrics:meter:generated") 3
               // Register the 'attempt' meter via @Metered in Service.java
               .bean("service") 4
               .filter(header(Exchange.REDELIVERED))
                   .log(LoggingLevel.WARN, "Processed ${body} after ${header.CamelRedeliveryCounter} retries")
                   .setHeader(MicroProfileMetricsConstants.HEADER_METER_MARK, header(Exchange.REDELIVERY_COUNTER))
                   // Register the 'redelivery' meter
                   .to("microprofile-metrics:meter:redelivery") 5
               .end()
               .log("Successfully processed ${body}")
               // Register the 'success' meter
               .to("microprofile-metrics:meter:success"); 6
       }
   }

   1

   使用 Camel K modeline 来自动配置 Prometheus trait 和 Maven 依赖项

   2

   错误 ：指标与尚未处理的事件数对应的错误数

   3

   generated ：指标要处理的事件数量

   4

   attempt：指标为服务 bean 发出的调用数，以处理传入的事件

   5

   redelivery ：指标为处理事件的重试次数

   6

   success：指标成功处理的事件数

2. 根据需要，将 Camel MicroProfile 指标注释添加到任何实施文件中。以下示例显示了 Camel K 集成调用的服务 bean，该服务会生成随机故障：

   package com.redhat.integration;
   
   import java.util.Random;
   
   import org.apache.camel.Exchange;
   import org.apache.camel.RuntimeExchangeException;
   
   import org.eclipse.microprofile.metrics.Meter;
   import org.eclipse.microprofile.metrics.annotation.Metered;
   import org.eclipse.microprofile.metrics.annotation.Metric;
   
   import javax.inject.Named;
   import javax.enterprise.context.ApplicationScoped;
   
   @Named("service")
   @ApplicationScoped
   @io.quarkus.arc.Unremovable
   
   public class Service {
   
      //Register the attempt meter
      @Metered(absolute = true)
      public void attempt(Exchange exchange) { 1
         Random rand = new Random();
            if (rand.nextDouble() < 0.5) {
                throw new RuntimeExchangeException("Random failure", exchange); 2
         }
      }
    }

   1

   @Metered MicroProfile Metrics 注释声明量表，名称会根据指标方法名称自动生成，本例中为 attempt。

   2

   本例随机失败，以帮助生成指标的错误。

3. 按照 配置 Camel K 集成指标 中的步骤运行集成，并在 Prometheus 中查看自定义 Camel K 指标。

   在这种情况下，示例已在 Metrics.java 中使用 Camel K 模式行来自动配置 Prometheus 和 Service.java 所需的 Maven 依赖项。

流程

1. 创建一个以 operator 指标端点为目标的 PodMonitor 资源，以便 Prometheus 服务器可以提取 Operator 公开的指标。

   operator-pod-monitor.yaml

   apiVersion: monitoring.coreos.com/v1
   kind: PodMonitor
   metadata:
     name: camel-k-operator
     labels:
       app: "camel-k"
       camel.apache.org/component: operator
   spec:
     selector:
       matchLabels:
         app: "camel-k"
         camel.apache.org/component: operator
     podMetricsEndpoints:
       - port: metrics

2. 创建 PodMonitor 资源。

   oc apply -f operator-pod-monitor.yaml

1. 安装所需的 OpenShift operator
2. 创建 Kafka 实例
3. 创建 Kafka 主题

运行制作者集成

1. 创建示例制作者集成。这会每 10 秒填充一条消息。

   Sample SaslSSLKafkaProducer.java

   // kamel run --secret kafka-props SaslSSLKafkaProducer.java --dev
   // camel-k: language=java dependency=mvn:org.apache.camel.quarkus:camel-quarkus-kafka dependency=mvn:io.strimzi:kafka-oauth-client:0.7.1.redhat-00003
   
   import org.apache.camel.builder.RouteBuilder;
   import org.apache.camel.component.kafka.KafkaConstants;
   
   public class SaslSSLKafkaProducer extends RouteBuilder {
     @Override
     public void configure() throws Exception {
     log.info("About to start route: Timer -> Kafka ");
     from("timer:foo")
       .routeId("FromTimer2Kafka")
       .setBody()
         .simple("Message #${exchangeProperty.CamelTimerCounter}")
       .to("kafka:{{producer.topic}}")
       .log("Message correctly sent to the topic!");
     }
   }

2. 然后，运行集成过程。

   kamel run --secret kafka-props SaslSSLKafkaProducer.java --dev

   制作者将创建一个新消息并推送到主题，并记录一些信息。

   [2] 2021-05-06 08:48:11,854 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #1 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:11,854 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #3 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:11,973 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #5 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:12,970 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #7 - KafkaProducer[test]) Message correctly sent to the topic!
   [2] 2021-05-06 08:48:13,970 INFO  [FromTimer2Kafka] (Camel (camel-1) thread #9 - KafkaProducer[test]) Message correctly sent to the topic!

运行消费者集成

1. 创建消费者集成。

   Sample SaslSSLKafkaProducer.java

   // kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev
   // camel-k: language=java dependency=mvn:org.apache.camel.quarkus:camel-quarkus-kafka dependency=mvn:io.strimzi:kafka-oauth-client:0.7.1.redhat-00003
   
   import org.apache.camel.builder.RouteBuilder;
   
   public class SaslSSLKafkaConsumer extends RouteBuilder {
     @Override
     public void configure() throws Exception {
   	log.info("About to start route: Kafka -> Log ");
   	from("kafka:{{consumer.topic}}")
       .routeId("FromKafka2Log")
       .log("${body}");
     }
   }

2. 打开另一个 shell，并使用以下命令运行消费者集成：

   kamel run --secret kafka-props SaslSSLKafkaConsumer.java --dev

   消费者将开始记录主题中找到的事件：

   [1] 2021-05-06 08:51:08,991 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #8
   [1] 2021-05-06 08:51:10,065 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #9
   [1] 2021-05-06 08:51:10,991 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #10
   [1] 2021-05-06 08:51:11,991 INFO  [FromKafka2Log] (Camel (camel-1) thread #0 - KafkaConsumer[test]) Message #11