Tanzu Application PlatformにSpring Cloud Data FlowのPre-packagedなStreamアプリをデプロイする

️⚠️ 本記事の内容はVMwareによってサポートされていません。記事の内容で生じた問題については自己責任で対応し、 VMwareサポート窓口には問い合わせないでください。

Spring Cloud Data Flow用に提供されているPre-packaged ApplicationsをTanzu Application Platform上にデプロイし、 Spring Cloud Data Flowのようなストリーム処理を実現してみます。

Pre-packaged Appはメッセージブローカー(RabbitMQ or Kafka)とパッケージ形式(Maven Repo or Docker Image)の組み合わせの4パターンで提供されています。

RabbitMQ / Maven https://dataflow.spring.io/rabbitmq-maven-latest
RabbitMQ / Docker https://dataflow.spring.io/rabbitmq-docker-latest
Kafka / Maven https://dataflow.spring.io/kafka-maven-latest
Kafka / Docker https://dataflow.spring.io/kafka-docker-latest

今回はメッセージブローカーにはRabbitMQを使用します。パッケージ形式としてはTAPでは Maven Repo も Docker Image も対応していますが、以下の理由からMaven Repo (Jarファイル) を使用します。

Docker Image(3.2.x系)はJibでビルドされておりspring-cloud-bindingsが内容されていないため、Service Binidngによるプロパティの自動設定が効かない
pre-builtなDocker Imageと異なり、Tanzu Build ServiceによるイメージのBuildpackやStackのアップデートが見込める

Bitnami Servicesを使用してRabbitMQサービスの作成
time | logなストリームのデプロイ
http | filter | logなストリームのデプロイ
http | jdbcなストリームのデプロイ

Bitnami Servicesを使用してRabbitMQサービスの作成

まずはRabbitMQを立ち上げます。本記事ではRabbitMQの作成にBitnami Servicesを使用します (要 TAP 1.5以上)。本記事の内容自体はサービスを素早く利用するためにBitnami Servicesを使用しますが、Service Bindingが使用できれば必ずしもBitnami Servicesを使用する必要はありません。

ℹ️ 本記事の内容をBitnami Servicesを使わずにRabbit MQをバインドしたい場合はこちらの記事を参照してください。

TAPから提供されているサービス一覧を次のコマンドで確認できます。

$ tanzu service class list
  NAME                  DESCRIPTION            
  mysql-unmanaged       MySQL by Bitnami       
  postgresql-unmanaged  PostgreSQL by Bitnami  
  rabbitmq-unmanaged    RabbitMQ by Bitnami    
  redis-unmanaged       Redis by Bitnami

rabbitmq-unmanagedクラスを使用して、次のようにClassClaimリソースを作成します。

tanzu service class-claim create hello-rabbitmq --class rabbitmq-unmanaged -n demo

次のコマンドで進捗を確認し、StatusのReadyがTrueになればOKです。

$ tanzu services class-claims get hello-rabbitmq --namespace demo             
Name: hello-rabbitmq
Namespace: demo
Claim Reference: services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq
Class Reference: 
  Name: rabbitmq-unmanaged
Parameters: None
Status: 
  Ready: True
  Claimed Resource: 
    Name: a65e8572-f3a8-4564-81fc-39c1702a05df
    Namespace: demo
    Group: 
    Version: v1
    Kind: Secret

ClassClaimはkubectlコマンドでも次のように確認できます。

$ kubectl get -n demo classclaims                             
NAME             READY   REASON   CLASSREF
hello-rabbitmq   True    Ready    rabbitmq-unmanaged

作成されたRabbitMQのPodは次のように確認できます。デフォルトではランダムなnamespace上に作成されます。

RESOURCE_NAME=$(kubectl get -n demo classclaims hello-rabbitmq -ojsonpath='{.status.provisionedResourceRef.name}')
kubectl get pod -n ${RESOURCE_NAME}

出力結果

NAME                     READY   STATUS    RESTARTS   AGE
hello-rabbitmq-qggk9-0   1/1     Running   0          6m

`time | log`なストリームのデプロイ

まずは次のHello World的なストリームをデプロイします。

Spring Cloud Data FlowのDSLで書くと次のようになります。

tictock=time | log

Time Sourceのデプロイ

まずはTime SourceのWorkloadを作成します。先ほど作成したRabbitMQをBindします。 spring.cloud.stream.bindings.output.destinationに宛先を指定します。ここに指定した値がRabbitMQのExchangeになります。

tanzu apps workload apply ticktock-time \
  --app ticktock-time \
  --maven-group org.springframework.cloud.stream.app \
  --maven-artifact time-source-rabbit \
  --maven-version 3.2.1 \
  --env spring.cloud.stream.bindings.output.destination=ticktock.time \
  --build-env BP_JVM_VERSION=17 \
  --type web \
  --annotation autoscaling.knative.dev/minScale=1 \
  --service-ref hello-rabbitmq=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq \
  -n demo \
  -y

sternを使うとログが見やすいです。

stern -n demo ticktock-time

次のような結果が得られればTime Sourceデプロイが完了です。

$ tanzu apps workload get ticktock-time --namespace demo 
📡 Overview
   name:        ticktock-time
   type:        web
   namespace:   demo

📦 Supply Chain
   name:   source-to-url

   NAME               READY   HEALTHY   UPDATED   RESOURCE
   source-provider    True    True      3m36s     mavenartifacts.source.apps.tanzu.vmware.com/ticktock-time
   image-provider     True    True      2m46s     images.kpack.io/ticktock-time
   config-provider    True    True      2m34s     podintents.conventions.carto.run/ticktock-time
   app-config         True    True      2m34s     configmaps/ticktock-time
   service-bindings   True    True      2m34s     configmaps/ticktock-time-with-claims
   api-descriptors    True    True      2m34s     configmaps/ticktock-time-with-api-descriptors
   config-writer      True    True      2m22s     runnables.carto.run/ticktock-time-config-writer

🚚 Delivery
   name:   delivery-basic

   NAME              READY   HEALTHY   UPDATED   RESOURCE
   source-provider   True    True      96s       imagerepositories.source.apps.tanzu.vmware.com/ticktock-time-delivery
   deployer          True    True      92s       apps.kappctrl.k14s.io/ticktock-time

💬 Messages
   No messages found.

🔁 Services
   CLAIM            NAME             KIND         API VERSION
   hello-rabbitmq   hello-rabbitmq   ClassClaim   services.apps.tanzu.vmware.com/v1alpha1

🛶 Pods
   NAME                                             READY   STATUS      RESTARTS   AGE
   ticktock-time-00001-deployment-65c6bf6dc4-vxfdk  2/2     Running     0          93s
   ticktock-time-build-1-build-pod                  0/1     Completed   0          3m36s
   ticktock-time-config-writer-7629m-pod            0/1     Completed   0          2m31s

🚢 Knative Services
   NAME           READY   URL
   ticktock-time   Ready   https://ticktock-time.demo.tap-happy-titmouse.tapsandbox.com

To see logs: "tanzu apps workload tail ticktock-time --namespace demo --timestamp --since 1h"

アプリのログは次のように出力されます。

$ stern -n demo ticktock-time -c workload

+ ticktock-time-00001-deployment-65c6bf6dc4-vxfdk › workload
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Setting Active Processor Count to 4
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Calculating JVM memory based on 14931448K available memory
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload For more information on this calculation, see https://paketo.io/docs/reference/java-reference/#memory-calculator
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Calculated JVM Memory Configuration: -XX:MaxDirectMemorySize=10M -Xmx14297292K -XX:MaxMetaspaceSize=122155K -XX:ReservedCodeCacheSize=240M -Xss1M (Total Memory: 14931448K, Thread Count: 250, Loaded Class Count: 19153, Headroom: 0%)
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Enabling Java Native Memory Tracking
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Adding 124 container CA certificates to JVM truststore
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Spring Cloud Bindings Enabled
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Picked up JAVA_TOOL_OPTIONS: -Dmanagement.endpoint.health.probes.add-additional-paths="true" -Dmanagement.health.probes.enabled="true" -Dserver.port="8080" -Dserver.shutdown.grace-period="24s" -Djava.security.properties=/layers/paketo-buildpacks_bellsoft-liberica/java-security-properties/java-security.properties -XX:+ExitOnOutOfMemoryError -XX:ActiveProcessorCount=4 -XX:MaxDirectMemorySize=10M -Xmx14297292K -XX:MaxMetaspaceSize=122155K -XX:ReservedCodeCacheSize=240M -Xss1M -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -Dorg.springframework.cloud.bindings.boot.enable=true
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:49.696  INFO [,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload   .   ____          _            __ _ _
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload   '  |____| .__|_| |_|_| |_\__, | / / / /
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload  =========|_|==============|___/=/_/_/_/
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload  :: Spring Boot ::                (v2.6.8)
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:50.640  INFO [time-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Fetching config from server at : http://localhost:8888
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:50.823  INFO [time-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Connect Timeout Exception on Url - http://localhost:8888. Will be trying the next url if available
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:50.823  WARN [time-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Could not locate PropertySource: I/O error on GET request for "http://localhost:8888/time-source/default": Connection refused; nested exception is java.net.ConnectException: Connection refused
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:50.830  INFO [time-source,,] 1 --- [           main] .c.s.a.t.s.r.TimeSourceRabbitApplication : No active profile set, falling back to 1 default profile: "default"
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:50.860  INFO [time-source,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:52.931  INFO [time-source,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:52.949  INFO [time-source,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:53.046  INFO [time-source,,] 1 --- [           main] o.s.cloud.context.scope.GenericScope     : BeanFactory id=cdb66a08-aadb-3227-8d4c-394f6beb28fc
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:53.539  INFO [time-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:53.578  INFO [time-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:53.658  INFO [time-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration' of type [org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:54.319  INFO [time-source,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:54.341  INFO [time-source,,] 1 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:54.342  INFO [time-source,,] 1 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.63]
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:54.511  INFO [time-source,,] 1 --- [           main] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:54.513  INFO [time-source,,] 1 --- [           main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 3652 ms
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:58.997  INFO [time-source,,] 1 --- [           main] o.s.s.web.DefaultSecurityFilterChain     : Will secure any request with [org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter@160cf225, org.springframework.security.web.context.SecurityContextPersistenceFilter@3bf41cd0, org.springframework.security.web.header.HeaderWriterFilter@33845e3d, org.springframework.security.web.csrf.CsrfFilter@656672fb, org.springframework.security.web.authentication.logout.LogoutFilter@51f009ef, org.springframework.security.web.authentication.UsernamePasswordAuthenticationFilter@63d75087, org.springframework.security.web.authentication.ui.DefaultLoginPageGeneratingFilter@7d4e424e, org.springframework.security.web.authentication.ui.DefaultLogoutPageGeneratingFilter@6a262980, org.springframework.security.web.authentication.www.BasicAuthenticationFilter@3435a4e5, org.springframework.security.web.savedrequest.RequestCacheAwareFilter@c4ca86a, org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter@52463255, org.springframework.security.web.authentication.AnonymousAuthenticationFilter@10230657, org.springframework.security.web.session.SessionManagementFilter@66851aa5, org.springframework.security.web.access.ExceptionTranslationFilter@71634e64, org.springframework.security.web.access.intercept.FilterSecurityInterceptor@245dbd09]
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.312  INFO [time-source,,] 1 --- [           main] ctiveUserDetailsServiceAutoConfiguration : 
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload Using generated security password: 77d3d612-d7cf-4af4-91f0-4f5180ea59da
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.676  INFO [time-source,,] 1 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 3 endpoint(s) beneath base path '/actuator'
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.872  INFO [time-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.873  INFO [time-source,,] 1 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application-1.errorChannel' has 1 subscriber(s).
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.874  INFO [time-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.875  INFO [time-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {router} as a subscriber to the 'timeSupplier_integrationflow.channel#0' channel
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.875  INFO [time-source,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.timeSupplier_integrationflow.channel#0' has 1 subscriber(s).
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.875  INFO [time-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'timeSupplier_integrationflow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:01:59.877  INFO [time-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Creating binder: rabbit
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.007  INFO [time-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Caching the binder: rabbit
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.007  INFO [time-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.103  INFO [time-source,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.187  INFO [time-source,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#6b030101:0/SimpleConnection@4cddc3d9 [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 56842]
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.250  INFO [time-source,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.output' has 1 subscriber(s).
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.257  INFO [time-source,,] 1 --- [           main] o.s.i.e.SourcePollingChannelAdapter      : started bean 'output_spca'
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.282  INFO [time-source,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.302  INFO [time-source,c4e05f87eb21af15,9aa51a21600eb1f1] 1 --- [   scheduling-1] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.328  INFO [time-source,c4e05f87eb21af15,9aa51a21600eb1f1] 1 --- [   scheduling-1] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory.publisher#37889e3a:0/SimpleConnection@6401c261 [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 56856]
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.342  INFO [time-source,,] 1 --- [           main] .c.s.a.t.s.r.TimeSourceRabbitApplication : Started TimeSourceRabbitApplication in 11.413 seconds (JVM running for 12.134)
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.401  INFO [time-source,679a4d45ea63e4e9,679a4d45ea63e4e9] 1 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.401  INFO [time-source,679a4d45ea63e4e9,679a4d45ea63e4e9] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
ticktock-time-00001-deployment-65c6bf6dc4-vxfdk workload 2023-05-17 07:02:00.405  INFO [time-source,679a4d45ea63e4e9,679a4d45ea63e4e9] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 4 ms

Log Sinkのデプロイ

次にLog SinkのWorkloadを作成します。spring.cloud.stream.bindings.input.destinationに受け取りたいSourceからの宛先(RabbitMQのExchange)を指定します。 spring.cloud.stream.bindings.input.groupにConsumer Group名を指定します。メッセージは同じConsumer Group内では(少なくとも)1インスタンスが受信します。

tanzu apps workload apply ticktock-log \
  --app ticktock-log \
  --maven-group org.springframework.cloud.stream.app \
  --maven-artifact log-sink-rabbit \
  --maven-version 3.2.1 \
  --env spring.cloud.stream.bindings.input.destination=ticktock.time \
  --env spring.cloud.stream.bindings.input.group=log \
  --build-env BP_JVM_VERSION=17 \
  --type web \
  --annotation autoscaling.knative.dev/minScale=1 \
  --service-ref hello-rabbitmq=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq \
  -n demo \
  -y

sternを使うとログが見やすいです。

stern -n demo ticktock-log

次のような結果が得られればLog Sinkのデプロイが完了です。

$ tanzu apps workload get ticktock-log --namespace demo                                                         
📡 Overview
   name:        ticktock-log
   type:        web
   namespace:   demo

📦 Supply Chain
   name:   source-to-url

   NAME               READY   HEALTHY   UPDATED   RESOURCE
   source-provider    True    True      3m58s     mavenartifacts.source.apps.tanzu.vmware.com/ticktock-log
   image-provider     True    True      2m41s     images.kpack.io/ticktock-log
   config-provider    True    True      2m27s     podintents.conventions.carto.run/ticktock-log
   app-config         True    True      2m27s     configmaps/ticktock-log
   service-bindings   True    True      2m27s     configmaps/ticktock-log-with-claims
   api-descriptors    True    True      2m26s     configmaps/ticktock-log-with-api-descriptors
   config-writer      True    True      2m14s     runnables.carto.run/ticktock-log-config-writer

🚚 Delivery
   name:   delivery-basic

   NAME              READY   HEALTHY   UPDATED   RESOURCE
   source-provider   True    True      119s      imagerepositories.source.apps.tanzu.vmware.com/ticktock-log-delivery
   deployer          True    True      111s      apps.kappctrl.k14s.io/ticktock-log

💬 Messages
   No messages found.

🔁 Services
   CLAIM            NAME             KIND         API VERSION
   hello-rabbitmq   hello-rabbitmq   ClassClaim   services.apps.tanzu.vmware.com/v1alpha1

🛶 Pods
   NAME                                            READY   STATUS      RESTARTS   AGE
   ticktock-log-00001-deployment-85ff54c4f8-w59zk  2/2     Running     0          115s
   ticktock-log-build-1-build-pod                  0/1     Completed   0          3m59s
   ticktock-log-config-writer-dmm7j-pod            0/1     Completed   0          2m24s

🚢 Knative Services
   NAME          READY   URL
   ticktock-log   Ready   https://ticktock-log.demo.tap-happy-titmouse.tapsandbox.com

To see logs: "tanzu apps workload tail ticktock-log --namespace demo --timestamp --since 1h"

アプリのログは次のように出力されます。Time Sourceが毎秒生成する時刻がLog Sinkで出力されていることがわかります。

$ stern -n demo ticktock-log -c workload               
+ ticktock-log-00001-deployment-85ff54c4f8-w59zkp › workload
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Setting Active Processor Count to 4
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Calculating JVM memory based on 14594260K available memory
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload For more information on this calculation, see https://paketo.io/docs/reference/java-reference/#memory-calculator
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Calculated JVM Memory Configuration: -XX:MaxDirectMemorySize=10M -Xmx13960823K -XX:MaxMetaspaceSize=121436K -XX:ReservedCodeCacheSize=240M -Xss1M (Total Memory: 14594260K, Thread Count: 250, Loaded Class Count: 19026, Headroom: 0%)
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Enabling Java Native Memory Tracking
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Adding 124 container CA certificates to JVM truststore
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Spring Cloud Bindings Enabled
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Picked up JAVA_TOOL_OPTIONS: -Dmanagement.endpoint.health.probes.add-additional-paths="true" -Dmanagement.health.probes.enabled="true" -Dserver.port="8080" -Dserver.shutdown.grace-period="24s" -Djava.security.properties=/layers/paketo-buildpacks_bellsoft-liberica/java-security-properties/java-security.properties -XX:+ExitOnOutOfMemoryError -XX:ActiveProcessorCount=4 -XX:MaxDirectMemorySize=10M -Xmx13960823K -XX:MaxMetaspaceSize=121436K -XX:ReservedCodeCacheSize=240M -Xss1M -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -Dorg.springframework.cloud.bindings.boot.enable=true
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:30.426  INFO [,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload   .   ____          _            __ _ _
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload   '  |____| .__|_| |_|_| |_\__, | / / / /
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload  =========|_|==============|___/=/_/_/_/
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload  :: Spring Boot ::                (v2.6.8)
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:31.234  INFO [log-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Fetching config from server at : http://localhost:8888
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:31.481  INFO [log-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Connect Timeout Exception on Url - http://localhost:8888. Will be trying the next url if available
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:31.482  WARN [log-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Could not locate PropertySource: I/O error on GET request for "http://localhost:8888/log-sink/default": Connection refused; nested exception is java.net.ConnectException: Connection refused
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:31.492  INFO [log-sink,,] 1 --- [           main] o.s.c.s.a.l.s.r.LogSinkRabbitApplication : No active profile set, falling back to 1 default profile: "default"
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:31.517  INFO [log-sink,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:33.729  INFO [log-sink,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:33.748  INFO [log-sink,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:33.844  INFO [log-sink,,] 1 --- [           main] o.s.cloud.context.scope.GenericScope     : BeanFactory id=8c49a73b-c9c8-3e5f-aaa9-0ad955efcb64
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:34.323  INFO [log-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:34.362  INFO [log-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:34.446  INFO [log-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration' of type [org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:35.137  INFO [log-sink,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:35.157  INFO [log-sink,,] 1 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:35.158  INFO [log-sink,,] 1 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.63]
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:35.300  INFO [log-sink,,] 1 --- [           main] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:35.301  INFO [log-sink,,] 1 --- [           main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 3781 ms
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:39.332  INFO [log-sink,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.input' has 1 subscriber(s).
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:39.870  INFO [log-sink,,] 1 --- [           main] o.s.s.web.DefaultSecurityFilterChain     : Will secure any request with [org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter@1d4531fa, org.springframework.security.web.context.SecurityContextPersistenceFilter@691a5c3a, org.springframework.security.web.header.HeaderWriterFilter@4e7151b3, org.springframework.security.web.csrf.CsrfFilter@73905dff, org.springframework.security.web.authentication.logout.LogoutFilter@29b0c169, org.springframework.security.web.authentication.UsernamePasswordAuthenticationFilter@3a00237d, org.springframework.security.web.authentication.ui.DefaultLoginPageGeneratingFilter@1afabf06, org.springframework.security.web.authentication.ui.DefaultLogoutPageGeneratingFilter@68cb8e52, org.springframework.security.web.authentication.www.BasicAuthenticationFilter@70a24463, org.springframework.security.web.savedrequest.RequestCacheAwareFilter@77648321, org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter@1db9c2cf, org.springframework.security.web.authentication.AnonymousAuthenticationFilter@3c2f310c, org.springframework.security.web.session.SessionManagementFilter@4e481512, org.springframework.security.web.access.ExceptionTranslationFilter@1701beb3, org.springframework.security.web.access.intercept.FilterSecurityInterceptor@46c475ba]
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.206  INFO [log-sink,,] 1 --- [           main] ctiveUserDetailsServiceAutoConfiguration : 
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload Using generated security password: 3f4c5114-0af2-421e-bdbe-ea9fce506dda
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.550  INFO [log-sink,,] 1 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 3 endpoint(s) beneath base path '/actuator'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.798  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.798  INFO [log-sink,,] 1 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application-1.errorChannel' has 1 subscriber(s).
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.799  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.799  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {service-activator} as a subscriber to the 'logConsumerFlow.channel#0' channel
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.799  INFO [log-sink,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.logConsumerFlow.channel#0' has 1 subscriber(s).
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.799  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'logConsumerFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'org.springframework.cloud.fn.consumer.log.LogConsumerConfiguration'; from source: 'bean method logConsumerFlow'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.799  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {bridge} as a subscriber to the 'logConsumerFlow.channel#2' channel
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.800  INFO [log-sink,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.logConsumerFlow.channel#2' has 1 subscriber(s).
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.800  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'logConsumerFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#1'; defined in: 'org.springframework.cloud.fn.consumer.log.LogConsumerConfiguration'; from source: 'bean method logConsumerFlow'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.801  INFO [log-sink,,] 1 --- [           main] ProxyFactoryBean$MethodInvocationGateway : started bean 'logConsumer#accept(Object)'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.801  INFO [log-sink,,] 1 --- [           main] .s.i.g.AnnotationGatewayProxyFactoryBean : started bean 'logConsumer'; defined in: 'org.springframework.cloud.fn.consumer.log.LogConsumerConfiguration'; from source: 'bean method logConsumerFlow'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.802  INFO [log-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Creating binder: rabbit
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.963  INFO [log-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Caching the binder: rabbit
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:40.964  INFO [log-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.084  INFO [log-sink,,] 1 --- [           main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: ticktock.time.log, bound to: ticktock.time
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.091  INFO [log-sink,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.187  INFO [log-sink,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#27dfd12b:0/SimpleConnection@46ef7340 [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 51486]
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.355  INFO [log-sink,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'ticktock.time.log.errors' has 1 subscriber(s).
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.357  INFO [log-sink,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'ticktock.time.log.errors' has 2 subscriber(s).
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.393  INFO [log-sink,,] 1 --- [           main] o.s.i.a.i.AmqpInboundChannelAdapter      : started bean 'inbound.ticktock.time.log'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.449  INFO [log-sink,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.527  INFO [log-sink,,] 1 --- [           main] o.s.c.s.a.l.s.r.LogSinkRabbitApplication : Started LogSinkRabbitApplication in 11.945 seconds (JVM running for 12.727)
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.659  INFO [log-sink,7f036fea0a27ac6e,7f036fea0a27ac6e] 1 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.659  INFO [log-sink,7f036fea0a27ac6e,7f036fea0a27ac6e] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.670  INFO [log-sink,7f036fea0a27ac6e,7f036fea0a27ac6e] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 8 ms
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:41.921  INFO [log-sink,feb5c123a83af154,fbe696da99b67149] 1 --- [tock.time.log-1] log-sink                                 : 05/17/23 07:05:41
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:42.818  INFO [log-sink,70be32383eb1315c,749f8d48f965939c] 1 --- [tock.time.log-1] log-sink                                 : 05/17/23 07:05:42
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:43.820  INFO [log-sink,dd581773f0fde253,eb8645497c3766cb] 1 --- [tock.time.log-1] log-sink                                 : 05/17/23 07:05:43
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:44.821  INFO [log-sink,260a6f55c61f7b16,b1bcbf52790db500] 1 --- [tock.time.log-1] log-sink                                 : 05/17/23 07:05:44
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:45.822  INFO [log-sink,c211574ec43a8a81,5d84c3cca43e0289] 1 --- [tock.time.log-1] log-sink                                 : 05/17/23 07:05:45
ticktock-log-00001-deployment-85ff54c4f8-w59zkp workload 2023-05-17 07:05:46.824  INFO [log-sink,a8368ae9ea72bdad,9149ac1975051d9d] 1 --- [tock.time.log-1] log-sink                                 : 05/17/23 07:05:46
...

RabbitMQ Management画面にアクセス

Bitnami ServicesでプロビジョンされたRabbitMQのManagement画面にアクセスしてみます。

次のコマンドでRabbitMQの認証情報を参照できます。

SECRET_NAME=$(kubectl get -n demo classclaims hello-rabbitmq -ojsonpath='{.status.binding.name}')
kubectl get secret -n demo ${SECRET_NAME} -ojson | jq '.data | map_values(@base64d)'

出力結果

{
  "host": "10.35.246.240",
  "password": "gznm7jS63pGiEEsbwdt71uwRjEq7zItR",
  "port": "5672",
  "provider": "bitnami",
  "type": "rabbitmq",
  "username": "rabbitmq"
}

次のコマンドでRabbitMQ Managemenをポートフォワードします。

RESOURCE_NAME=$(kubectl get -n demo classclaims hello-rabbitmq -ojsonpath='{.status.provisionedResourceRef.name}')
kubectl port-forward -n ${RESOURCE_NAME} svc/${RESOURCE_NAME} 15672:15672

http://localhost:15672 でRabbitMQ Management画面にアクセスできます。

"Exchange"タブをクリックしてExchange一覧画面へ遷移すると、ticktock.time Exchangeが作成されていることがわかります。

ticktock.time Exchangeをクリックすると詳細を見ることができます。毎秒メッセージが送られていることがわかります。

"Queue"タブをクリックしてQueue一覧画面へ遷移すると、ticktock.time.log Queueが作成されていることがわかります。

ticktock.time.log Queueをクリックすると詳細を見ることができます。毎秒メッセージを処理していることがわかります。

確認したらWorkloadを全て削除します。

kubectl delete workload -n demo --all

`http | filter | log`なストリームのデプロイ

次にProcessorを含む次のようなストリームをデプロイします。

DSLで定義すると次のようなストリームを作成します。Filter ProcessorではPayloadのうちorigin.countryがJPであるものだけを通過させます。

hello-stream=http | filter --expression=#jsonPath(payload,'$.origin.country').equals('JP') | log

HTTP Sourceのデプロイ

まずはHTTP SourceのWorkloadを作成します。

tanzu apps workload apply hello-stream-http \
  --app hello-stream-http \
  --maven-group org.springframework.cloud.stream.app \
  --maven-artifact http-source-rabbit \
  --maven-version 3.2.1 \
  --env spring.cloud.stream.bindings.output.destination=hello-stream.http \
  --build-env BP_JVM_VERSION=17 \
  --type web \
  --annotation autoscaling.knative.dev/minScale=1 \
  --service-ref hello-rabbitmq=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq \
  -n demo \
  -y

Filter Processorのデプロイ

次にFilter ProcessorのWorkloadを作成します。

tanzu apps workload apply hello-stream-filter \
  --app hello-stream-filter \
  --maven-group org.springframework.cloud.stream.app \
  --maven-artifact filter-processor-rabbit \
  --maven-version 3.2.1 \
  --env spring.cloud.stream.function.bindings.filterFunction-in-0=input \
  --env spring.cloud.stream.function.bindings.filterFunction-out-0=output \
  --env spring.cloud.stream.bindings.input.destination=hello-stream.http \
  --env spring.cloud.stream.bindings.input.group=filter \
  --env spring.cloud.stream.bindings.output.destination=hello-stream.filter \
  --env filter.function.expression="#jsonPath(payload,'$.origin.country').equals('JP')" \
  --build-env BP_JVM_VERSION=17 \
  --type web \
  --annotation autoscaling.knative.dev/minScale=1 \
  --service-ref hello-rabbitmq=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq \
  -n demo \
  -y

spring.cloud.stream.function.bindingsに指定する関数名は以下の各関数のREADMEから読み取れます。 https://github.com/spring-cloud/stream-applications/tree/main/functions

Log Sinkのデプロイ

最後にLog SinkのWorkloadを作成します。

tanzu apps workload apply hello-stream-log \
  --app hello-stream-log \
  --maven-group org.springframework.cloud.stream.app \
  --maven-artifact log-sink-rabbit \
  --maven-version 3.2.1 \
  --env spring.cloud.stream.bindings.input.destination=hello-stream.filter \
  --env spring.cloud.stream.bindings.input.group=log \
  --build-env BP_JVM_VERSION=17 \
  --type web \
  --annotation autoscaling.knative.dev/minScale=1 \
  --service-ref hello-rabbitmq=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq \
  -n demo \
  -y

3つのWorkloadがREADYになるまで待ちます。

$ kubectl get workload -n demo 
NAME                  SOURCE   SUPPLYCHAIN     READY   REASON   AGE
hello-stream-filter            source-to-url   True    Ready    3m28s
hello-stream-http              source-to-url   True    Ready    3m34s
hello-stream-log               source-to-url   True    Ready    3m22s

各アプリのログは次のように出力されます。

$ stern -n demo hello-stream -c workload
+ hello-stream-http-00001-deployment-8547b44865-68cm8 › workload
+ hello-stream-log-00001-deployment-86859cf948-fp6fq › workload
+ hello-stream-filter-00001-deployment-8768f9d9f-jdv9f › workload
hello-stream-http-00001-deployment-8547b44865-68cm8 workload Setting Active Processor Count to 4
hello-stream-http-00001-deployment-8547b44865-68cm8 workload Calculating JVM memory based on 14662088K available memory
hello-stream-http-00001-deployment-8547b44865-68cm8 workload For more information on this calculation, see https://paketo.io/docs/reference/java-reference/#memory-calculator
hello-stream-http-00001-deployment-8547b44865-68cm8 workload Calculated JVM Memory Configuration: -XX:MaxDirectMemorySize=10M -Xmx14028504K -XX:MaxMetaspaceSize=121583K -XX:ReservedCodeCacheSize=240M -Xss1M (Total Memory: 14662088K, Thread Count: 250, Loaded Class Count: 19052, Headroom: 0%)
hello-stream-http-00001-deployment-8547b44865-68cm8 workload Enabling Java Native Memory Tracking
hello-stream-http-00001-deployment-8547b44865-68cm8 workload Adding 124 container CA certificates to JVM truststore
hello-stream-http-00001-deployment-8547b44865-68cm8 workload Spring Cloud Bindings Enabled
hello-stream-http-00001-deployment-8547b44865-68cm8 workload Picked up JAVA_TOOL_OPTIONS: -Dmanagement.endpoint.health.probes.add-additional-paths="true" -Dmanagement.health.probes.enabled="true" -Dserver.port="8080" -Dserver.shutdown.grace-period="24s" -Djava.security.properties=/layers/paketo-buildpacks_bellsoft-liberica/java-security-properties/java-security.properties -XX:+ExitOnOutOfMemoryError -XX:ActiveProcessorCount=4 -XX:MaxDirectMemorySize=10M -Xmx14028504K -XX:MaxMetaspaceSize=121583K -XX:ReservedCodeCacheSize=240M -Xss1M -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -Dorg.springframework.cloud.bindings.boot.enable=true
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:45.961  INFO [,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 
hello-stream-http-00001-deployment-8547b44865-68cm8 workload   .   ____          _            __ _ _
hello-stream-http-00001-deployment-8547b44865-68cm8 workload  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
hello-stream-http-00001-deployment-8547b44865-68cm8 workload ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
hello-stream-http-00001-deployment-8547b44865-68cm8 workload  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
hello-stream-http-00001-deployment-8547b44865-68cm8 workload   '  |____| .__|_| |_|_| |_\__, | / / / /
hello-stream-http-00001-deployment-8547b44865-68cm8 workload  =========|_|==============|___/=/_/_/_/
hello-stream-http-00001-deployment-8547b44865-68cm8 workload  :: Spring Boot ::                (v2.6.8)
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:46.866  INFO [http-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Fetching config from server at : http://localhost:8888
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:47.059  INFO [http-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Connect Timeout Exception on Url - http://localhost:8888. Will be trying the next url if available
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:47.060  WARN [http-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Could not locate PropertySource: I/O error on GET request for "http://localhost:8888/http-source/default": Connection refused; nested exception is java.net.ConnectException: Connection refused
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:47.069  INFO [http-source,,] 1 --- [           main] .c.s.a.h.s.r.HttpSourceRabbitApplication : No active profile set, falling back to 1 default profile: "default"
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:47.096  INFO [http-source,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:49.153  INFO [http-source,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:49.215  INFO [http-source,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:49.329  INFO [http-source,,] 1 --- [           main] o.s.cloud.context.scope.GenericScope     : BeanFactory id=40874edf-cf9d-3979-933e-25888624a014
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:49.849  INFO [http-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:49.870  INFO [http-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:49.926  INFO [http-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration' of type [org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:50.423  INFO [http-source,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:50.450  INFO [http-source,,] 1 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:50.451  INFO [http-source,,] 1 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.63]
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:51.467  WARN [http-source,,] 1 --- [           main] o.s.i.w.inbound.WebFluxInboundEndpoint   : The 'requestPayloadType' attribute will have no relevance for one of the specified HTTP methods '[GET, HEAD, OPTIONS]'
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:53.496  INFO [http-source,,] 1 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 3 endpoint(s) beneath base path '/actuator'
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.077  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.077  INFO [http-source,,] 1 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application-1.errorChannel' has 1 subscriber(s).
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.078  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.078  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {header-enricher} as a subscriber to the 'httpSupplierFlow.channel#0' channel
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.079  INFO [http-source,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.httpSupplierFlow.channel#0' has 1 subscriber(s).
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.079  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'httpSupplierFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'org.springframework.cloud.fn.supplier.http.HttpSupplierConfiguration'; from source: 'bean method httpSupplierFlow'
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.107  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.ReactiveStreamsConsumer   : started bean 'httpSupplier_integrationflow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.109  INFO [http-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Creating binder: rabbit
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.335  INFO [http-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Caching the binder: rabbit
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.335  INFO [http-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.473  INFO [http-source,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.570  INFO [http-source,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#a1cb94:0/SimpleConnection@6f87a7c7 [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 57020]
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.650  INFO [http-source,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.output' has 1 subscriber(s).
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.658  INFO [http-source,,] 1 --- [           main] o.s.i.w.inbound.WebFluxInboundEndpoint   : started bean 'httpSupplierFlow.webflux:inbound-channel-adapter#0'; defined in: 'org.springframework.cloud.fn.supplier.http.HttpSupplierConfiguration'; from source: 'bean method httpSupplierFlow'
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.726  INFO [http-source,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
hello-stream-http-00001-deployment-8547b44865-68cm8 workload 2023-05-17 07:12:55.870  INFO [http-source,,] 1 --- [           main] .c.s.a.h.s.r.HttpSourceRabbitApplication : Started HttpSourceRabbitApplication in 10.745 seconds (JVM running for 11.442)

hello-stream-log-00001-deployment-86859cf948-fp6fq workload Setting Active Processor Count to 4
hello-stream-log-00001-deployment-86859cf948-fp6fq workload Calculating JVM memory based on 14313456K available memory
hello-stream-log-00001-deployment-86859cf948-fp6fq workload For more information on this calculation, see https://paketo.io/docs/reference/java-reference/#memory-calculator
hello-stream-log-00001-deployment-86859cf948-fp6fq workload Calculated JVM Memory Configuration: -XX:MaxDirectMemorySize=10M -Xmx13680019K -XX:MaxMetaspaceSize=121436K -XX:ReservedCodeCacheSize=240M -Xss1M (Total Memory: 14313456K, Thread Count: 250, Loaded Class Count: 19026, Headroom: 0%)
hello-stream-log-00001-deployment-86859cf948-fp6fq workload Enabling Java Native Memory Tracking
hello-stream-log-00001-deployment-86859cf948-fp6fq workload Adding 124 container CA certificates to JVM truststore
hello-stream-log-00001-deployment-86859cf948-fp6fq workload Spring Cloud Bindings Enabled
hello-stream-log-00001-deployment-86859cf948-fp6fq workload Picked up JAVA_TOOL_OPTIONS: -Dmanagement.endpoint.health.probes.add-additional-paths="true" -Dmanagement.health.probes.enabled="true" -Dserver.port="8080" -Dserver.shutdown.grace-period="24s" -Djava.security.properties=/layers/paketo-buildpacks_bellsoft-liberica/java-security-properties/java-security.properties -XX:+ExitOnOutOfMemoryError -XX:ActiveProcessorCount=4 -XX:MaxDirectMemorySize=10M -Xmx13680019K -XX:MaxMetaspaceSize=121436K -XX:ReservedCodeCacheSize=240M -Xss1M -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -Dorg.springframework.cloud.bindings.boot.enable=true
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:12:59.813  INFO [,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 
hello-stream-log-00001-deployment-86859cf948-fp6fq workload   .   ____          _            __ _ _
hello-stream-log-00001-deployment-86859cf948-fp6fq workload  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
hello-stream-log-00001-deployment-86859cf948-fp6fq workload ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
hello-stream-log-00001-deployment-86859cf948-fp6fq workload  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
hello-stream-log-00001-deployment-86859cf948-fp6fq workload   '  |____| .__|_| |_|_| |_\__, | / / / /
hello-stream-log-00001-deployment-86859cf948-fp6fq workload  =========|_|==============|___/=/_/_/_/
hello-stream-log-00001-deployment-86859cf948-fp6fq workload  :: Spring Boot ::                (v2.6.8)
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:01.097  INFO [log-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Fetching config from server at : http://localhost:8888
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:01.324  INFO [log-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Connect Timeout Exception on Url - http://localhost:8888. Will be trying the next url if available
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:01.325  WARN [log-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Could not locate PropertySource: I/O error on GET request for "http://localhost:8888/log-sink/default": Connection refused; nested exception is java.net.ConnectException: Connection refused
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:01.336  INFO [log-sink,,] 1 --- [           main] o.s.c.s.a.l.s.r.LogSinkRabbitApplication : No active profile set, falling back to 1 default profile: "default"
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:01.375  INFO [log-sink,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:05.068  INFO [log-sink,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:05.092  INFO [log-sink,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:05.187  INFO [log-sink,,] 1 --- [           main] o.s.cloud.context.scope.GenericScope     : BeanFactory id=8c49a73b-c9c8-3e5f-aaa9-0ad955efcb64
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:05.692  INFO [log-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:05.727  INFO [log-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:05.811  INFO [log-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration' of type [org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:06.508  INFO [log-sink,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:06.526  INFO [log-sink,,] 1 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:06.527  INFO [log-sink,,] 1 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.63]
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:06.673  INFO [log-sink,,] 1 --- [           main] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:06.675  INFO [log-sink,,] 1 --- [           main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 5298 ms
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:10.898  INFO [log-sink,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.input' has 1 subscriber(s).
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:11.553  INFO [log-sink,,] 1 --- [           main] o.s.s.web.DefaultSecurityFilterChain     : Will secure any request with [org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter@1107c465, org.springframework.security.web.context.SecurityContextPersistenceFilter@4e7151b3, org.springframework.security.web.header.HeaderWriterFilter@22c4354d, org.springframework.security.web.csrf.CsrfFilter@1a73f3d1, org.springframework.security.web.authentication.logout.LogoutFilter@21b2579d, org.springframework.security.web.authentication.UsernamePasswordAuthenticationFilter@1a53ac0c, org.springframework.security.web.authentication.ui.DefaultLoginPageGeneratingFilter@2c51c756, org.springframework.security.web.authentication.ui.DefaultLogoutPageGeneratingFilter@64df8422, org.springframework.security.web.authentication.www.BasicAuthenticationFilter@59c00722, org.springframework.security.web.savedrequest.RequestCacheAwareFilter@30b97fcf, org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter@3d213a2b, org.springframework.security.web.authentication.AnonymousAuthenticationFilter@2b30b627, org.springframework.security.web.session.SessionManagementFilter@7f584d0c, org.springframework.security.web.access.ExceptionTranslationFilter@64921450, org.springframework.security.web.access.intercept.FilterSecurityInterceptor@23e573e]
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:11.902  INFO [log-sink,,] 1 --- [           main] ctiveUserDetailsServiceAutoConfiguration : 
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 
hello-stream-log-00001-deployment-86859cf948-fp6fq workload Using generated security password: a40f2563-e570-419a-a9c4-80dfba039e67
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.182  INFO [log-sink,,] 1 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 3 endpoint(s) beneath base path '/actuator'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.370  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.372  INFO [log-sink,,] 1 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application-1.errorChannel' has 1 subscriber(s).
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.374  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.375  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {service-activator} as a subscriber to the 'logConsumerFlow.channel#0' channel
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.376  INFO [log-sink,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.logConsumerFlow.channel#0' has 1 subscriber(s).
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.376  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'logConsumerFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'org.springframework.cloud.fn.consumer.log.LogConsumerConfiguration'; from source: 'bean method logConsumerFlow'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.376  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {bridge} as a subscriber to the 'logConsumerFlow.channel#2' channel
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.376  INFO [log-sink,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.logConsumerFlow.channel#2' has 1 subscriber(s).
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.376  INFO [log-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'logConsumerFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#1'; defined in: 'org.springframework.cloud.fn.consumer.log.LogConsumerConfiguration'; from source: 'bean method logConsumerFlow'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.378  INFO [log-sink,,] 1 --- [           main] ProxyFactoryBean$MethodInvocationGateway : started bean 'logConsumer#accept(Object)'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.378  INFO [log-sink,,] 1 --- [           main] .s.i.g.AnnotationGatewayProxyFactoryBean : started bean 'logConsumer'; defined in: 'org.springframework.cloud.fn.consumer.log.LogConsumerConfiguration'; from source: 'bean method logConsumerFlow'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.379  INFO [log-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Creating binder: rabbit
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.575  INFO [log-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Caching the binder: rabbit
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.575  INFO [log-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.690  INFO [log-sink,,] 1 --- [           main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: hello-stream.filter.log, bound to: hello-stream.filter
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.696  INFO [log-sink,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.785  INFO [log-sink,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#56872fcb:0/SimpleConnection@401d83ce [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 48978]
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.883  INFO [log-sink,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'hello-stream.filter.log.errors' has 1 subscriber(s).
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.885  INFO [log-sink,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'hello-stream.filter.log.errors' has 2 subscriber(s).
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.914  INFO [log-sink,,] 1 --- [           main] o.s.i.a.i.AmqpInboundChannelAdapter      : started bean 'inbound.hello-stream.filter.log'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:12.953  INFO [log-sink,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:13.013  INFO [log-sink,,] 1 --- [           main] o.s.c.s.a.l.s.r.LogSinkRabbitApplication : Started LogSinkRabbitApplication in 14.491 seconds (JVM running for 15.931)
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:13.066  INFO [log-sink,ed1d3ba632aaf416,ed1d3ba632aaf416] 1 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:13.067  INFO [log-sink,ed1d3ba632aaf416,ed1d3ba632aaf416] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:13:13.076  INFO [log-sink,ed1d3ba632aaf416,ed1d3ba632aaf416] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 9 ms

hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Setting Active Processor Count to 4
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Calculating JVM memory based on 14897720K available memory
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload For more information on this calculation, see https://paketo.io/docs/reference/java-reference/#memory-calculator
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Calculated JVM Memory Configuration: -XX:MaxDirectMemorySize=10M -Xmx14264289K -XX:MaxMetaspaceSize=121430K -XX:ReservedCodeCacheSize=240M -Xss1M (Total Memory: 14897720K, Thread Count: 250, Loaded Class Count: 19025, Headroom: 0%)
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Enabling Java Native Memory Tracking
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Adding 124 container CA certificates to JVM truststore
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Spring Cloud Bindings Enabled
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Picked up JAVA_TOOL_OPTIONS: -Dmanagement.endpoint.health.probes.add-additional-paths="true" -Dmanagement.health.probes.enabled="true" -Dserver.port="8080" -Dserver.shutdown.grace-period="24s" -Djava.security.properties=/layers/paketo-buildpacks_bellsoft-liberica/java-security-properties/java-security.properties -XX:+ExitOnOutOfMemoryError -XX:ActiveProcessorCount=4 -XX:MaxDirectMemorySize=10M -Xmx14264289K -XX:MaxMetaspaceSize=121430K -XX:ReservedCodeCacheSize=240M -Xss1M -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -Dorg.springframework.cloud.bindings.boot.enable=true
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:50.511  INFO [,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload   .   ____          _            __ _ _
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload   '  |____| .__|_| |_|_| |_\__, | / / / /
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload  =========|_|==============|___/=/_/_/_/
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload  :: Spring Boot ::                (v2.6.8)
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:51.375  INFO [filter-processor,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Fetching config from server at : http://localhost:8888
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:51.591  INFO [filter-processor,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Connect Timeout Exception on Url - http://localhost:8888. Will be trying the next url if available
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:51.592  WARN [filter-processor,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Could not locate PropertySource: I/O error on GET request for "http://localhost:8888/filter-processor/default": Connection refused; nested exception is java.net.ConnectException: Connection refused
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:51.609  INFO [filter-processor,,] 1 --- [           main] a.f.p.r.FilterProcessorRabbitApplication : No active profile set, falling back to 1 default profile: "default"
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:51.648  INFO [filter-processor,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:53.747  INFO [filter-processor,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:53.774  INFO [filter-processor,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:53.873  INFO [filter-processor,,] 1 --- [           main] o.s.cloud.context.scope.GenericScope     : BeanFactory id=1867de95-ac1e-3ae9-b48d-e1f45e52f458
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:54.440  INFO [filter-processor,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:54.471  INFO [filter-processor,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:54.540  INFO [filter-processor,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration' of type [org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:55.342  INFO [filter-processor,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:55.357  INFO [filter-processor,,] 1 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:55.360  INFO [filter-processor,,] 1 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.63]
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:55.486  INFO [filter-processor,,] 1 --- [           main] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:55.486  INFO [filter-processor,,] 1 --- [           main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 3838 ms
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:58.996  INFO [filter-processor,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Creating binder: rabbit
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:59.232  INFO [filter-processor,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Caching the binder: rabbit
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:59.232  INFO [filter-processor,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:59.354  INFO [filter-processor,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.input' has 1 subscriber(s).
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:11:59.809  INFO [filter-processor,,] 1 --- [           main] o.s.s.web.DefaultSecurityFilterChain     : Will secure any request with [org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter@3407ded1, org.springframework.security.web.context.SecurityContextPersistenceFilter@fdf029a, org.springframework.security.web.header.HeaderWriterFilter@96c840a, org.springframework.security.web.csrf.CsrfFilter@18fa5af6, org.springframework.security.web.authentication.logout.LogoutFilter@7c680fe1, org.springframework.security.web.authentication.UsernamePasswordAuthenticationFilter@4113d9ab, org.springframework.security.web.authentication.ui.DefaultLoginPageGeneratingFilter@35451ba6, org.springframework.security.web.authentication.ui.DefaultLogoutPageGeneratingFilter@63af52a6, org.springframework.security.web.authentication.www.BasicAuthenticationFilter@61f08aa9, org.springframework.security.web.savedrequest.RequestCacheAwareFilter@517d9cd5, org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter@1c528f2f, org.springframework.security.web.authentication.AnonymousAuthenticationFilter@10ae002e, org.springframework.security.web.session.SessionManagementFilter@6e0f259e, org.springframework.security.web.access.ExceptionTranslationFilter@5eae392d, org.springframework.security.web.access.intercept.FilterSecurityInterceptor@5c20505f]
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.106  INFO [filter-processor,,] 1 --- [           main] ctiveUserDetailsServiceAutoConfiguration : 
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload Using generated security password: 3b1fc69a-9736-4cc9-954a-9f83d4efcdec
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.372  INFO [filter-processor,,] 1 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 3 endpoint(s) beneath base path '/actuator'
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.623  INFO [filter-processor,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.624  INFO [filter-processor,,] 1 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application-1.errorChannel' has 1 subscriber(s).
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.625  INFO [filter-processor,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.627  INFO [filter-processor,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.797  INFO [filter-processor,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.882  INFO [filter-processor,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#391d28ea:0/SimpleConnection@574134ba [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 60656]
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.941  INFO [filter-processor,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.output' has 1 subscriber(s).
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:00.945  INFO [filter-processor,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.030  INFO [filter-processor,,] 1 --- [           main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: hello-stream.http.filter, bound to: hello-stream.http
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.076  INFO [filter-processor,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'hello-stream.http.filter.errors' has 1 subscriber(s).
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.076  INFO [filter-processor,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'hello-stream.http.filter.errors' has 2 subscriber(s).
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.099  INFO [filter-processor,,] 1 --- [           main] o.s.i.a.i.AmqpInboundChannelAdapter      : started bean 'inbound.hello-stream.http.filter'
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.120  INFO [filter-processor,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.154  INFO [filter-processor,,] 1 --- [           main] a.f.p.r.FilterProcessorRabbitApplication : Started FilterProcessorRabbitApplication in 11.62 seconds (JVM running for 12.27)
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.260  INFO [filter-processor,4dd0c356cbff9b52,4dd0c356cbff9b52] 1 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.261  INFO [filter-processor,4dd0c356cbff9b52,4dd0c356cbff9b52] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
hello-stream-filter-00001-deployment-8768f9d9f-jdv9f workload 2023-05-17 07:12:01.266  INFO [filter-processor,4dd0c356cbff9b52,4dd0c356cbff9b52] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 3 ms

HTTP Sourceに次のようにリクエストを送ります。

SOURCE_URL=$(kubectl get ksvc -n demo hello-stream-http -ojsonpath='{.status.url}')

curl -kv ${SOURCE_URL} -H "Content-Type: application/json" -d "{\"origin\": {\"id\": $RANDOM, \"country\": \"US\"}}" 
curl -kv ${SOURCE_URL} -H "Content-Type: application/json" -d "{\"origin\": {\"id\": $RANDOM, \"country\": \"JP\"}}"
curl -kv ${SOURCE_URL} -H "Content-Type: application/json" -d "{\"origin\": {\"id\": $RANDOM, \"country\": \"US\"}}" 
curl -kv ${SOURCE_URL} -H "Content-Type: application/json" -d "{\"origin\": {\"id\": $RANDOM, \"country\": \"JP\"}}"

Log Sinkのログを確認します。origin.countryがJPであるものだけが出力されていることがわかります。

$ stern -n demo hello-stream-log -c workload                                                   
+ hello-stream-log-00001-deployment-86859cf948-fp6fq › workload
...
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:14:52.530  INFO [log-sink,110a0088c0cfc99b,d723854aee752b0d] 1 --- [am.filter.log-1] log-sink                                 : {"origin": {"id": 3449, "country": "JP"}}
hello-stream-log-00001-deployment-86859cf948-fp6fq workload 2023-05-17 07:15:01.502  INFO [log-sink,d3aa3a018297e377,37e4687a2a4a5cb0] 1 --- [am.filter.log-1] log-sink                                 : {"origin": {"id": 1475, "country": "JP"}}

RabbitMQ Managementを見ると、以下のExchangeができています。

hello-stream.http ... HTTP Sourceからの宛先
hello-stream.filter ... Filter Sourceからの宛先

また、以下のQueueもできています。

hello-stream.http.filter ... HTTP SourceとFilter Processorを繋ぐキュー
hello-stream.filter.log ... Filter ProcessorとLog Sinkを繋ぐキュー

確認したらWorkloadを全て削除します。

kubectl delete workload -n demo --all

`http | jdbc`なストリームのデプロイ

今度はデータベースに書き込む次にようなストリームをデプロイします。

DSLで定義すると次のようなストリームを作成します。

http-to-db=http | jdbc --table-name=usage --columns=first_name:firstName,last_name:lastName,minutes:minutes,data_usage:dataUsage --spring.datasource.driver-class-name=org.postgresql.Driver --spring.datasource.url="jdbc:postgresql://demo-db:5432/demo" --spring.datasource.username=demo --spring.datasource.password=demo

Bitnami Servicesを使用してPostgreSQLサービスの作成

書き込み先のPostgreSQLもBitnami Servicesを使用して手軽に作成します。RabbitMQと同様に、Service Bindingが使用できれば必ずしもBitnami Servicesを使用する必要はありません。

postgresql-unmanagedクラスを使用して、次のようにClassClaimリソースを作成します。

tanzu service class-claim create hello-db --class postgresql-unmanaged --parameter storageGB=1 -n demo

次のコマンドで進捗を確認し、StatusのReadyがTrueになればOKです。

$ tanzu services class-claims get hello-db --namespace demo   
Name: hello-db
Namespace: demo
Claim Reference: services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-db
Class Reference: 
  Name: postgresql-unmanaged
Parameters: 
  storageGB: 1
Status: 
  Ready: True
  Claimed Resource: 
    Name: 246f8eb5-2784-4c53-9629-3395cc1801d8
    Namespace: demo
    Group: 
    Version: v1
    Kind: Secret

HTTP Sourceのデプロイ

まずはHTTP SourceのWorkloadを作成します。

tanzu apps workload apply http-to-db-http \
  --app http-to-db-http \
  --maven-group org.springframework.cloud.stream.app \
  --maven-artifact http-source-rabbit \
  --maven-version 3.2.1 \
  --env spring.cloud.stream.bindings.output.destination=http-to-db.http \
  --build-env BP_JVM_VERSION=17 \
  --type web \
  --annotation autoscaling.knative.dev/minScale=1 \
  --service-ref hello-rabbitmq=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq \
  -n demo \
  -y

JDBC Sinkのデプロイ

次にLog SinkのWorkloadを作成します。先ほど作成したPostgreSQLもBindします。

tanzu apps workload apply http-to-db-db \
  --app http-to-db-db \
  --maven-group org.springframework.cloud.stream.app \
  --maven-artifact jdbc-sink-rabbit \
  --maven-version 3.2.1 \
  --env spring.cloud.stream.bindings.input.destination=http-to-db.http \
  --env spring.cloud.stream.bindings.input.group=jdbc \
  --env jdbc.consumer.table-name=usage \
  --env jdbc.consumer.columns=first_name:firstName,last_name:lastName,minutes:minutes,data_usage:dataUsage \
  --env spring.sql.init.mode=always \
  --env spring.sql.init.schema-locations=https://gist.githubusercontent.com/making/09b7d571f9f74323e8007f58f317d7cc/raw/a93c55a60ecfd8c4f101f92752f3cdebbdb9f194/schema.sql \
  --build-env BP_JVM_VERSION=17 \
  --type web \
  --annotation autoscaling.knative.dev/minScale=1 \
  --service-ref hello-rabbitmq=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-rabbitmq \
  --service-ref hello-db=services.apps.tanzu.vmware.com/v1alpha1:ClassClaim:hello-db \
  -n demo \
  -y

各アプリのログは次のように出力されます。

$ stern -n demo http-to-db -c workload
+ http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 › workload
+ http-to-db-db-00001-deployment-6c8565f6d8-ccfwc › workload
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload Setting Active Processor Count to 4
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload Calculating JVM memory based on 14926864K available memory
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload For more information on this calculation, see https://paketo.io/docs/reference/java-reference/#memory-calculator
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload Calculated JVM Memory Configuration: -XX:MaxDirectMemorySize=10M -Xmx14293280K -XX:MaxMetaspaceSize=121583K -XX:ReservedCodeCacheSize=240M -Xss1M (Total Memory: 14926864K, Thread Count: 250, Loaded Class Count: 19052, Headroom: 0%)
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload Enabling Java Native Memory Tracking
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload Adding 124 container CA certificates to JVM truststore
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload Spring Cloud Bindings Enabled
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload Picked up JAVA_TOOL_OPTIONS: -Dmanagement.endpoint.health.probes.add-additional-paths="true" -Dmanagement.health.probes.enabled="true" -Dserver.port="8080" -Dserver.shutdown.grace-period="24s" -Djava.security.properties=/layers/paketo-buildpacks_bellsoft-liberica/java-security-properties/java-security.properties -XX:+ExitOnOutOfMemoryError -XX:ActiveProcessorCount=4 -XX:MaxDirectMemorySize=10M -Xmx14293280K -XX:MaxMetaspaceSize=121583K -XX:ReservedCodeCacheSize=240M -Xss1M -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -Dorg.springframework.cloud.bindings.boot.enable=true
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:47.053  INFO [,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload   .   ____          _            __ _ _
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload   '  |____| .__|_| |_|_| |_\__, | / / / /
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload  =========|_|==============|___/=/_/_/_/
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload  :: Spring Boot ::                (v2.6.8)
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:48.087  INFO [http-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Fetching config from server at : http://localhost:8888
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:48.306  INFO [http-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Connect Timeout Exception on Url - http://localhost:8888. Will be trying the next url if available
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:48.307  WARN [http-source,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Could not locate PropertySource: I/O error on GET request for "http://localhost:8888/http-source/default": Connection refused; nested exception is java.net.ConnectException: Connection refused
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:48.315  INFO [http-source,,] 1 --- [           main] .c.s.a.h.s.r.HttpSourceRabbitApplication : No active profile set, falling back to 1 default profile: "default"
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:48.346  INFO [http-source,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:50.678  INFO [http-source,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:50.707  INFO [http-source,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:50.827  INFO [http-source,,] 1 --- [           main] o.s.cloud.context.scope.GenericScope     : BeanFactory id=40874edf-cf9d-3979-933e-25888624a014
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:51.447  INFO [http-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:51.475  INFO [http-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:51.536  INFO [http-source,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration' of type [org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:52.058  INFO [http-source,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:52.081  INFO [http-source,,] 1 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:52.081  INFO [http-source,,] 1 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.63]
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:53.180  WARN [http-source,,] 1 --- [           main] o.s.i.w.inbound.WebFluxInboundEndpoint   : The 'requestPayloadType' attribute will have no relevance for one of the specified HTTP methods '[GET, HEAD, OPTIONS]'
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:55.008  INFO [http-source,,] 1 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 3 endpoint(s) beneath base path '/actuator'
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.893  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.893  INFO [http-source,,] 1 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application-1.errorChannel' has 1 subscriber(s).
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.894  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.894  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {header-enricher} as a subscriber to the 'httpSupplierFlow.channel#0' channel
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.895  INFO [http-source,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.httpSupplierFlow.channel#0' has 1 subscriber(s).
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.895  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'httpSupplierFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'org.springframework.cloud.fn.supplier.http.HttpSupplierConfiguration'; from source: 'bean method httpSupplierFlow'
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.921  INFO [http-source,,] 1 --- [           main] o.s.i.endpoint.ReactiveStreamsConsumer   : started bean 'httpSupplier_integrationflow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:56.923  INFO [http-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Creating binder: rabbit
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.079  INFO [http-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Caching the binder: rabbit
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.079  INFO [http-source,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.185  INFO [http-source,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.273  INFO [http-source,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#46d148bd:0/SimpleConnection@857f1a7 [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 40150]
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.353  INFO [http-source,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.output' has 1 subscriber(s).
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.356  INFO [http-source,,] 1 --- [           main] o.s.i.w.inbound.WebFluxInboundEndpoint   : started bean 'httpSupplierFlow.webflux:inbound-channel-adapter#0'; defined in: 'org.springframework.cloud.fn.supplier.http.HttpSupplierConfiguration'; from source: 'bean method httpSupplierFlow'
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.423  INFO [http-source,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:48:57.478  INFO [http-source,,] 1 --- [           main] .c.s.a.h.s.r.HttpSourceRabbitApplication : Started HttpSourceRabbitApplication in 11.314 seconds (JVM running for 12.348)
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:52:54.219  INFO [http-source,f486126ea2ff7b8c,66deade5b159a26d] 1 --- [oundedElastic-1] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
http-to-db-http-00001-deployment-7cddd4f8cd-jxtl2 workload 2023-05-18 00:52:54.228  INFO [http-source,f486126ea2ff7b8c,66deade5b159a26d] 1 --- [oundedElastic-1] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory.publisher#6136cb80:0/SimpleConnection@fe2b827 [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 60638]

http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Setting Active Processor Count to 4
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Calculating JVM memory based on 14832336K available memory
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload For more information on this calculation, see https://paketo.io/docs/reference/java-reference/#memory-calculator
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Calculated JVM Memory Configuration: -XX:MaxDirectMemorySize=10M -Xmx14191961K -XX:MaxMetaspaceSize=128374K -XX:ReservedCodeCacheSize=240M -Xss1M (Total Memory: 14832336K, Thread Count: 250, Loaded Class Count: 20251, Headroom: 0%)
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Enabling Java Native Memory Tracking
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Adding 124 container CA certificates to JVM truststore
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Spring Cloud Bindings Enabled
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Picked up JAVA_TOOL_OPTIONS: -Dmanagement.endpoint.health.probes.add-additional-paths="true" -Dmanagement.health.probes.enabled="true" -Dserver.port="8080" -Dserver.shutdown.grace-period="24s" -Djava.security.properties=/layers/paketo-buildpacks_bellsoft-liberica/java-security-properties/java-security.properties -XX:+ExitOnOutOfMemoryError -XX:ActiveProcessorCount=4 -XX:MaxDirectMemorySize=10M -Xmx14191961K -XX:MaxMetaspaceSize=128374K -XX:ReservedCodeCacheSize=240M -Xss1M -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -Dorg.springframework.cloud.bindings.boot.enable=true
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:09.166  INFO [,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload   .   ____          _            __ _ _
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload  /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload  \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload   '  |____| .__|_| |_|_| |_\__, | / / / /
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload  =========|_|==============|___/=/_/_/_/
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload  :: Spring Boot ::                (v2.6.8)
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:10.008  INFO [jdbc-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Fetching config from server at : http://localhost:8888
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:10.183  INFO [jdbc-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Connect Timeout Exception on Url - http://localhost:8888. Will be trying the next url if available
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:10.184  WARN [jdbc-sink,,] 1 --- [           main] c.c.c.ConfigServicePropertySourceLocator : Could not locate PropertySource: I/O error on GET request for "http://localhost:8888/jdbc-sink/default": Connection refused; nested exception is java.net.ConnectException: Connection refused
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:10.194  INFO [jdbc-sink,,] 1 --- [           main] .s.c.s.a.j.s.r.JdbcSinkRabbitApplication : No active profile set, falling back to 1 default profile: "default"
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:10.221  INFO [jdbc-sink,,] 1 --- [           main] .BindingSpecificEnvironmentPostProcessor : Creating binding-specific PropertySource from Kubernetes Service Bindings
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:12.898  INFO [jdbc-sink,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:12.931  INFO [jdbc-sink,,] 1 --- [           main] faultConfiguringBeanFactoryPostProcessor : No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:13.028  INFO [jdbc-sink,,] 1 --- [           main] o.s.cloud.context.scope.GenericScope     : BeanFactory id=531a253e-e98f-3ff8-bc7e-876711036f30
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:13.712  INFO [jdbc-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:13.748  INFO [jdbc-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:13.830  INFO [jdbc-sink,,] 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration' of type [org.springframework.cloud.stream.app.postprocessor.ContentTypeHeaderBeanPostProcessorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:14.508  INFO [jdbc-sink,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:14.523  INFO [jdbc-sink,,] 1 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:14.524  INFO [jdbc-sink,,] 1 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.63]
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:14.663  INFO [jdbc-sink,,] 1 --- [           main] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:14.665  INFO [jdbc-sink,,] 1 --- [           main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 4442 ms
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:18.709  INFO [jdbc-sink,,] 1 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.input' has 1 subscriber(s).
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:18.792  INFO [jdbc-sink,,] 1 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Starting...
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:19.190  INFO [jdbc-sink,,] 1 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Start completed.
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:19.706  INFO [jdbc-sink,,] 1 --- [           main] o.s.s.web.DefaultSecurityFilterChain     : Will secure any request with [org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter@6eb332d1, org.springframework.security.web.context.SecurityContextPersistenceFilter@2e3b4394, org.springframework.security.web.header.HeaderWriterFilter@20b921be, org.springframework.security.web.csrf.CsrfFilter@6158113e, org.springframework.security.web.authentication.logout.LogoutFilter@6a4d57b7, org.springframework.security.web.authentication.UsernamePasswordAuthenticationFilter@56f19f36, org.springframework.security.web.authentication.ui.DefaultLoginPageGeneratingFilter@6b7a0f18, org.springframework.security.web.authentication.ui.DefaultLogoutPageGeneratingFilter@34d0c331, org.springframework.security.web.authentication.www.BasicAuthenticationFilter@3d9cba2, org.springframework.security.web.savedrequest.RequestCacheAwareFilter@1952ad9d, org.springframework.security.web.servletapi.SecurityContextHolderAwareRequestFilter@232be28d, org.springframework.security.web.authentication.AnonymousAuthenticationFilter@3958db82, org.springframework.security.web.session.SessionManagementFilter@12b66c89, org.springframework.security.web.access.ExceptionTranslationFilter@1715193f, org.springframework.security.web.access.intercept.FilterSecurityInterceptor@16c1345b]
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:20.571  INFO [jdbc-sink,,] 1 --- [           main] ctiveUserDetailsServiceAutoConfiguration : 
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload Using generated security password: 499e8f12-a72f-430d-aa6d-304191ebbc01
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.025  INFO [jdbc-sink,,] 1 --- [           main] o.s.b.a.e.web.EndpointLinksResolver      : Exposing 3 endpoint(s) beneath base path '/actuator'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.312  INFO [jdbc-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.312  INFO [jdbc-sink,,] 1 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'application-1.errorChannel' has 1 subscriber(s).
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.313  INFO [jdbc-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean '_org.springframework.integration.errorLogger'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.313  INFO [jdbc-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {jdbc:outbound-channel-adapter} as a subscriber to the 'jdbcConsumerFlow.channel#0' channel
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.314  INFO [jdbc-sink,,] 1 --- [           main] o.s.integration.channel.DirectChannel    : Channel 'application-1.jdbcConsumerFlow.channel#0' has 1 subscriber(s).
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.315  INFO [jdbc-sink,,] 1 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started bean 'jdbcConsumerFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'org.springframework.cloud.fn.consumer.jdbc.JdbcConsumerConfiguration'; from source: 'bean method jdbcConsumerFlow'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.317  INFO [jdbc-sink,,] 1 --- [           main] ProxyFactoryBean$MethodInvocationGateway : started bean 'jdbcConsumer#accept(Object)'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.317  INFO [jdbc-sink,,] 1 --- [           main] .s.i.g.AnnotationGatewayProxyFactoryBean : started bean 'jdbcConsumer'; defined in: 'org.springframework.cloud.fn.consumer.jdbc.JdbcConsumerConfiguration'; from source: 'bean method jdbcConsumerFlow'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.318  INFO [jdbc-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Creating binder: rabbit
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.474  INFO [jdbc-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Caching the binder: rabbit
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.475  INFO [jdbc-sink,,] 1 --- [           main] o.s.c.s.binder.DefaultBinderFactory      : Retrieving cached binder: rabbit
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.681  INFO [jdbc-sink,,] 1 --- [           main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: http-to-db.http.jdbc, bound to: http-to-db.http
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.687  INFO [jdbc-sink,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [10.35.246.240:5672]
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.804  INFO [jdbc-sink,,] 1 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#65b73689:0/SimpleConnection@7a80ebcd [delegate=amqp://rabbitmq@10.35.246.240:5672/, localPort= 49284]
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.900  INFO [jdbc-sink,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'http-to-db.http.jdbc.errors' has 1 subscriber(s).
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.909  INFO [jdbc-sink,,] 1 --- [           main] o.s.c.stream.binder.BinderErrorChannel   : Channel 'http-to-db.http.jdbc.errors' has 2 subscriber(s).
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.937  INFO [jdbc-sink,,] 1 --- [           main] o.s.i.a.i.AmqpInboundChannelAdapter      : started bean 'inbound.http-to-db.http.jdbc'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:21.960  INFO [jdbc-sink,,] 1 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:22.019  INFO [jdbc-sink,,] 1 --- [           main] .s.c.s.a.j.s.r.JdbcSinkRabbitApplication : Started JdbcSinkRabbitApplication in 13.698 seconds (JVM running for 14.503)
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:22.085  INFO [jdbc-sink,16c521117645066d,16c521117645066d] 1 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:22.085  INFO [jdbc-sink,16c521117645066d,16c521117645066d] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
http-to-db-db-00001-deployment-6c8565f6d8-ccfwc workload 2023-05-18 00:57:22.089  INFO [jdbc-sink,16c521117645066d,16c521117645066d] 1 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 4 ms

RabbitMQ Managementを見ると、以下のExchangeができています。

http-to-db.http ... HTTP Sourceからの宛先

また、以下のQueueもできています。

http-to-db.http.jdbc ... HTTP SourceとJDBC Sinkを繋ぐキュー

HTTP Sourceに次のようにリクエストを送ります。

SOURCE_URL=$(kubectl get ksvc -n demo http-to-db-http -ojsonpath='{.status.url}')

curl -kv ${SOURCE_URL} -H "Content-Type: application/json" -d "{\"firstName\": \"John\", \"lastName\": \"Doe\", \"minutes\": 500, \"dataUsage\": 1000}"

次のように、Bitnami Servicesで作成したPostgreSQLに対して、SQLを実行します。

DB_NAME=$(kubectl get -n demo classclaims hello-db -ojsonpath='{.status.provisionedResourceRef.name}')

kubectl exec -ti -n ${DB_NAME} ${DB_NAME}-0 -- bash -c 'PGPASSWORD=${POSTGRES_PASSWORD} /opt/bitnami/postgresql/bin/psql -U postgres ${POSTGRES_DB} -c "SELECT * FROM usage"'

次のような結果が返り、送信したデータがDBに格納されたことがわかります。

 id | first_name | last_name | minutes | data_usage |          created_at           
----+------------+-----------+---------+------------+-------------------------------
  1 | John       | Doe       |     500 |       1000 | 2023-05-18 01:09:54.096024+00
(1 row)

確認したらWorkloadを全て削除します。

kubectl delete workload -n demo --all

Tanzu Application PlatformにSpring Cloud StreamのPre-packagedなStreamアプリをデプロイできました。
Spring Cloud Data Flowでデプロイするのとは異なり、コンポーネント同士の結び付けは自分で必要があります。記事の内容ではストリームのインデックスの設定を行っていないので、Partitioning (Stateful Stream) は利用できない点に注意する必要があります。

Partitioningを行う場合は、ドキュメントにしたがって、
送信側にはspring.cloud.stream.bindings.<binding_name>.producer.partition-key-expressionやspring.cloud.stream.bindings.<binding_name>.producer.partition-countの設定を、
受信側にはspring.cloud.stream.bindings.<binding_name>.consumer.partitionedやspring.cloud.stream.instance-index、spring.cloud.stream.instance-countの設定を行う必要があります。
この辺りの設定をSpring Cloud Data Flowが行ってくれます。

とはいえ、このようなストリーム処理もTanzu Application PlatformのSupply Chainを利用して管理できるのは便利だと思います。

IK.AM