使用 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 特征，如下所示：

   $ 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 控制台中，在 Cursor 的 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

   生成的 ：指标要处理的事件数

   4

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

   5

   重新传送 ：指标为处理事件而重试的重试次数

   6

   成功 ：考虑成功处理的事件数量

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 注释声明了计量，并根据指标方法名称自动生成名称，在这种情况下，尝试。

   2

   这个示例没有随机失败，以帮助生成指标的错误。

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

   在本例中，示例已经使用 Metrics.java 中的 Camel K modeline 自动配置 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. 创建示例制作者集成。这会用每 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