Camel K를 사용하여 통합 개발 및 관리

프로세스

1. Camel K Operator가 OpenShift 클러스터에서 실행 중인지 확인합니다. 예를 들면 다음과 같습니다.

   Copy to Clipboard Toggle word wrap

   oc get pod

   Copy to Clipboard Toggle word wrap

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

2. kamel run 명령을 입력하여 OpenShift의 클라우드에서 통합을 실행합니다. 예를 들면 다음과 같습니다.

   Copy to Clipboard Toggle word wrap

   kamel run hello.camelk.yaml

   Copy to Clipboard Toggle word wrap

   integration "hello" created

3. kamel get 명령을 입력하여 통합 상태를 확인합니다.

   Copy to Clipboard Toggle word wrap

   kamel get

   Copy to Clipboard Toggle word wrap

   NAME   PHASE        KIT
   hello  Building Kit kit-bqatqib5t4kse5vukt40

4. kamel describe 명령을 입력하여 통합에 대한 자세한 정보를 출력합니다.

   Copy to Clipboard Toggle word wrap

   kamel describe integration hello

   Copy to Clipboard Toggle word wrap

   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. 로그를 stdout 에 출력하려면 kamel log 명령을 입력합니다.

   Copy to Clipboard Toggle word wrap

   kamel log hello

   Copy to Clipboard Toggle word wrap

   ...
   [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. OpenShift에 배포된 통합을 삭제하려면 kamel delete 를 입력합니다.

   Copy to Clipboard Toggle word wrap

   kamel delete hello

   Copy to Clipboard Toggle word wrap

   Integration hello deleted

프로세스

1. kamel run 명령을 입력하고 --property 옵션을 사용하여 로깅 수준을 지정합니다. 예를 들면 다음과 같습니다.

   Copy to Clipboard Toggle word wrap

   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 오브젝트가 있는지 확인합니다.

   Copy to Clipboard Toggle word wrap

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

2. 아직 존재하지 않는 경우 cluster-monitoring-config ConfigMap을 생성합니다.

   Copy to Clipboard Toggle word wrap

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

3. cluster-monitoring-config ConfigMap을 편집합니다.

   Copy to Clipboard Toggle word wrap

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

4. data:config.yaml: 에서 enableUserWorkload 를 true 로 설정합니다.

   Copy to Clipboard Toggle word wrap

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

프로세스

1. 다음 명령을 입력하여 Prometheus 특성을 활성화한 Camel K 통합을 실행합니다.

   Copy to Clipboard Toggle word wrap

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

   또는 다음과 같이 통합 플랫폼을 업데이트하여 Prometheus 특성을 전역적으로 한 번 활성화할 수 있습니다.

   Copy to Clipboard Toggle word wrap

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

2. Prometheus에서 Camel K 통합 지표의 모니터링을 확인합니다. 예를 들어 포함된 Prometheus의 경우 OpenShift 관리자 또는 개발자 웹 콘솔에서 모니터링 > 메트릭 을 선택합니다.
3. 확인할 Camel K 메트릭을 입력합니다. 예를 들어 관리자 콘솔에서 Cursor의 Metric 삽입 아래에 application_camel_context_uptime_seconds 를 입력하고 Run Queries 를 클릭합니다.
4. 쿼리 추가 를 클릭하여 추가 지표를 확인합니다.

프로세스

1. Camel MicroProfile Metrics 구성 요소 주석을 사용하여 Camel 통합 코드에 사용자 지정 지표를 등록합니다. 다음 예제에서는 Metrics.java 통합을 보여줍니다.

   Copy to Clipboard Toggle word wrap

   // 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 모드 라인을 사용하여 Prometheus 특성 및 Maven 종속성을 자동으로 구성

   2

   Error: 처리되지 않은 이벤트 수에 해당하는 오류 수를 측정

   3

   생성됨: 처리할 이벤트 수에 대한 메트릭

   4

   attempt: 들어오는 이벤트를 처리하기 위해 서비스 빈에 대한 호출 수를 측정

   5

   Redelivery: 이벤트를 처리하기 위한 재시도 횟수에 대한 지표

   6

   성공: 성공적으로 처리된 이벤트 수를 측정

2. 필요에 따라 Camel MicroProfile Metrics 주석을 구현 파일에 추가합니다. 다음 예제에서는 임의의 오류를 생성하는 Camel K 통합에서 호출하는 서비스 빈을 보여줍니다.

   Copy to Clipboard Toggle word wrap

   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 모드를 사용하여 Prometheus 및 Service.java 에 필요한 Maven 종속성을 자동으로 구성합니다.

프로세스

1. Prometheus 서버가 Operator에서 노출하는 메트릭을 스크랩할 수 있도록 Operator 지표 끝점을 대상으로 하는 PodMonitor 리소스를 생성합니다.

   operator-pod-monitor.yaml

   Copy to Clipboard Toggle word wrap

   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 리소스를 생성합니다.

   Copy to Clipboard Toggle word wrap

   oc apply -f operator-pod-monitor.yaml

1. 필요한 OpenShift Operator 설치
2. Kafka 인스턴스 생성
3. Kafka 주제 생성

생산자 통합 실행

1. 샘플 생산자 통합을 생성합니다. 이렇게 하면 주제가 10초마다 메시지로 채워집니다.

   Sample SaslSSLKafkaProducer.java

   Copy to Clipboard Toggle word wrap

   // 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. 그런 다음 절차 통합을 실행합니다.

   Copy to Clipboard Toggle word wrap

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

   생산자는 새 메시지를 생성하고 항목에 푸시하고 일부 정보를 기록합니다.

   Copy to Clipboard Toggle word wrap

   [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

   Copy to Clipboard Toggle word wrap

   // 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. 다른 쉘을 열고 명령을 사용하여 소비자 통합을 실행합니다.

   Copy to Clipboard Toggle word wrap

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

   소비자가 Topic에 있는 이벤트 로깅을 시작합니다.

   Copy to Clipboard Toggle word wrap

   [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