From b4114ac043e7d906806818a51702c290dabc7068 Mon Sep 17 00:00:00 2001 From: Darcy Date: Fri, 23 Apr 2021 11:29:35 +0800 Subject: [PATCH] enhancement agent kafka reporter plugin, add namespace concept (#6781) --- CHANGES.md | 2 + apm-sniffer/config/agent.config | 2 + .../core/kafka/KafkaConnectionStatus.java | 23 ++++ .../kafka/KafkaConnectionStatusListener.java | 23 ++++ .../KafkaContextManagerExtendService.java | 9 +- .../core/kafka/KafkaJVMMetricsSender.java | 31 +++--- .../agent/core/kafka/KafkaMeterSender.java | 18 +++- .../core/kafka/KafkaProducerManager.java | 100 +++++++++++++----- .../kafka/KafkaProfileSnapshotSender.java | 17 ++- .../core/kafka/KafkaReporterPluginConfig.java | 2 + .../KafkaServiceManagementServiceClient.java | 18 +++- .../kafka/KafkaTraceSegmentServiceClient.java | 17 ++- .../core/kafka/KafkaProducerManagerTest.java | 69 ++++++++++++ docs/en/setup/backend/backend-fetcher.md | 7 ++ .../setup/backend/configuration-vocabulary.md | 1 + .../setup/service-agent/java-agent/README.md | 1 + .../src/main/resources/application.yml | 1 + .../kafka/module/KafkaFetcherConfig.java | 2 + .../handler/AbstractKafkaHandler.java | 54 ++++++++++ .../provider/handler/JVMMetricsHandler.java | 15 +-- .../provider/handler/MeterServiceHandler.java | 14 +-- .../provider/handler/ProfileTaskHandler.java | 15 +-- .../handler/ServiceManagementHandler.java | 14 +-- .../provider/handler/TraceSegmentHandler.java | 17 +-- .../handler/AbstractKafkaHandlerTest.java | 86 +++++++++++++++ 25 files changed, 450 insertions(+), 108 deletions(-) create mode 100644 apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatus.java create mode 100644 apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatusListener.java create mode 100644 apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/test/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManagerTest.java create mode 100644 oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/AbstractKafkaHandler.java create mode 100644 oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/test/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/AbstractKafkaHandlerTest.java diff --git a/CHANGES.md b/CHANGES.md index 20e778874..e9bc8714f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -12,6 +12,8 @@ Release Notes. * Improve `GlobalIdGenerator` performance. * Add an agent plugin to support elasticsearch7. * Add `jsonrpc4j` agent plugin. +* new options to support multi skywalking cluster use same kafka cluster(plugin.kafka.namespace) +* resolve agent has no retries if connect kafka cluster failed when bootstrap * Add Seata in the component definition. Seata plugin hosts on Seata project. * Extended Kafka plugin to properly trace consumers that have topic partitions directly assigned. * Support print SkyWalking context to logs. diff --git a/apm-sniffer/config/agent.config b/apm-sniffer/config/agent.config index 6e3757032..1bfadde76 100644 --- a/apm-sniffer/config/agent.config +++ b/apm-sniffer/config/agent.config @@ -107,6 +107,8 @@ plugin.mount=${SW_MOUNT_FOLDERS:plugins,activations} # Kafka producer configuration # plugin.kafka.bootstrap_servers=${SW_KAFKA_BOOTSTRAP_SERVERS:localhost:9092} +# if you want to set namespace. please make sure the OAP server has set it in Kafka fetcher module +# plugin.kafka.namspacce=${SW_KAFKA_NAMESPACE:""} # Match spring bean with regex expression for classname # plugin.springannotation.classname_match_regex=${SW_SPRINGANNOTATION_CLASSNAME_MATCH_REGEX:} diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatus.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatus.java new file mode 100644 index 000000000..b1ce94821 --- /dev/null +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatus.java @@ -0,0 +1,23 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.agent.core.kafka; + +enum KafkaConnectionStatus { + CONNECTED, DISCONNECT +} diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatusListener.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatusListener.java new file mode 100644 index 000000000..ae26e9a46 --- /dev/null +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaConnectionStatusListener.java @@ -0,0 +1,23 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.agent.core.kafka; + +interface KafkaConnectionStatusListener { + void onStatusChanged(KafkaConnectionStatus status); +} diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaContextManagerExtendService.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaContextManagerExtendService.java index 8cab8e705..f5e812446 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaContextManagerExtendService.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaContextManagerExtendService.java @@ -19,6 +19,7 @@ package org.apache.skywalking.apm.agent.core.kafka; import org.apache.skywalking.apm.agent.core.boot.OverrideImplementor; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; import org.apache.skywalking.apm.agent.core.context.ContextManagerExtendService; import org.apache.skywalking.apm.agent.core.remote.GRPCChannelStatus; @@ -26,11 +27,15 @@ import org.apache.skywalking.apm.agent.core.remote.GRPCChannelStatus; * For compatible with {@link ContextManagerExtendService}, don't need to manage connection status by self. */ @OverrideImplementor(ContextManagerExtendService.class) -public class KafkaContextManagerExtendService extends ContextManagerExtendService { +public class KafkaContextManagerExtendService extends ContextManagerExtendService implements KafkaConnectionStatusListener { @Override public void prepare() { - statusChanged(GRPCChannelStatus.CONNECTED); + ServiceManager.INSTANCE.findService(KafkaProducerManager.class).addListener(this); } + @Override + public void onStatusChanged(KafkaConnectionStatus status) { + statusChanged(GRPCChannelStatus.CONNECTED); + } } diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaJVMMetricsSender.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaJVMMetricsSender.java index 6a399eb8a..ba67021bb 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaJVMMetricsSender.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaJVMMetricsSender.java @@ -22,6 +22,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; + import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.utils.Bytes; @@ -38,13 +39,11 @@ import org.apache.skywalking.apm.network.language.agent.v3.JVMMetricCollection; * A report to send JVM Metrics data to Kafka Broker. */ @OverrideImplementor(JVMMetricsSender.class) -public class KafkaJVMMetricsSender extends JVMMetricsSender { +public class KafkaJVMMetricsSender extends JVMMetricsSender implements KafkaConnectionStatusListener { private static final ILog LOGGER = LogManager.getLogger(KafkaJVMMetricsSender.class); private KafkaProducer producer; private String topic; - private BlockingQueue queue; - private volatile boolean running = false; @Override public void run() { @@ -52,7 +51,7 @@ public class KafkaJVMMetricsSender extends JVMMetricsSender { List buffer = new ArrayList<>(); queue.drainTo(buffer); - if (running) { + if (producer != null) { JVMMetricCollection metrics = JVMMetricCollection.newBuilder() .addAllMetrics(buffer) .setService(Config.Agent.SERVICE_NAME) @@ -61,15 +60,15 @@ public class KafkaJVMMetricsSender extends JVMMetricsSender { if (LOGGER.isDebugEnable()) { LOGGER.debug( - "JVM metrics reporting, topic: {}, key: {}, length: {}", topic, metrics.getServiceInstance(), - buffer.size() + "JVM metrics reporting, topic: {}, key: {}, length: {}", topic, metrics.getServiceInstance(), + buffer.size() ); } producer.send(new ProducerRecord<>( - topic, - metrics.getServiceInstance(), - Bytes.wrap(metrics.toByteArray()) + topic, + metrics.getServiceInstance(), + Bytes.wrap(metrics.toByteArray()) )); producer.flush(); } @@ -79,14 +78,13 @@ public class KafkaJVMMetricsSender extends JVMMetricsSender { @Override public void prepare() { queue = new LinkedBlockingQueue<>(Config.Jvm.BUFFER_SIZE); - topic = KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_METRICS; + KafkaProducerManager producerManager = ServiceManager.INSTANCE.findService(KafkaProducerManager.class); + producerManager.addListener(this); + topic = producerManager.formatTopicNameThenRegister(KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_METRICS); } @Override public void boot() { - producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); - running = true; - } @Override @@ -96,4 +94,11 @@ public class KafkaJVMMetricsSender extends JVMMetricsSender { queue.offer(metric); } } + + @Override + public void onStatusChanged(KafkaConnectionStatus status) { + if (status == KafkaConnectionStatus.CONNECTED) { + producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); + } + } } diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaMeterSender.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaMeterSender.java index 26e30f24f..c69abd2bc 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaMeterSender.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaMeterSender.java @@ -19,6 +19,7 @@ package org.apache.skywalking.apm.agent.core.kafka; import java.util.Map; + import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.utils.Bytes; @@ -37,7 +38,7 @@ import org.apache.skywalking.apm.network.language.agent.v3.MeterDataCollection; * A report to send Metrics data of meter system to Kafka Broker. */ @OverrideImplementor(MeterSender.class) -public class KafkaMeterSender extends MeterSender { +public class KafkaMeterSender extends MeterSender implements KafkaConnectionStatusListener { private static final ILog LOGGER = LogManager.getLogger(KafkaTraceSegmentServiceClient.class); private String topic; @@ -45,16 +46,20 @@ public class KafkaMeterSender extends MeterSender { @Override public void prepare() { - topic = KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_METER; + KafkaProducerManager producerManager = ServiceManager.INSTANCE.findService(KafkaProducerManager.class); + producerManager.addListener(this); + topic = producerManager.formatTopicNameThenRegister(KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_METER); } @Override public void boot() { - producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); } @Override public void send(Map meterMap, MeterService meterService) { + if (producer == null) { + return; + } MeterDataCollection.Builder builder = MeterDataCollection.newBuilder(); transform(meterMap, meterData -> { if (LOGGER.isDebugEnable()) { @@ -67,4 +72,11 @@ public class KafkaMeterSender extends MeterSender { producer.flush(); } + + @Override + public void onStatusChanged(KafkaConnectionStatus status) { + if (status == KafkaConnectionStatus.CONNECTED) { + producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); + } + } } diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java index 70ebcf6a1..efa1e138b 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java @@ -18,14 +18,19 @@ package org.apache.skywalking.apm.agent.core.kafka; -import java.util.Arrays; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; import java.util.Objects; import java.util.Properties; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; + import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.DescribeTopicsResult; import org.apache.kafka.clients.producer.KafkaProducer; @@ -35,8 +40,13 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.Bytes; import org.apache.skywalking.apm.agent.core.boot.BootService; import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; +import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; import org.apache.skywalking.apm.agent.core.logging.api.ILog; import org.apache.skywalking.apm.agent.core.logging.api.LogManager; +import org.apache.skywalking.apm.agent.core.remote.GRPCChannelManager; +import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; +import org.apache.skywalking.apm.util.StringUtil; /** * Configuring, initializing and holding a KafkaProducer instance for reporters. @@ -46,23 +56,53 @@ public class KafkaProducerManager implements BootService, Runnable { private static final ILog LOGGER = LogManager.getLogger(KafkaProducerManager.class); - private KafkaProducer producer; + private Set topics = new HashSet<>(); + private List listeners = new ArrayList<>(); + + private volatile KafkaProducer producer; + + private ScheduledFuture bootProducerFuture; @Override - public void prepare() throws Throwable { + public void prepare() { + } + + @Override + public void boot() { + bootProducerFuture = Executors.newSingleThreadScheduledExecutor( + new DefaultNamedThreadFactory("kafkaProducerInitThread") + ).scheduleAtFixedRate(new RunnableWithExceptionProtection( + this, + t -> LOGGER.error("unexpected exception.", t) + ), 0, 120, TimeUnit.SECONDS); + } + + String formatTopicNameThenRegister(String topic) { + String topicName = StringUtil.isBlank(KafkaReporterPluginConfig.Plugin.Kafka.NAMESPACE) ? topic + : KafkaReporterPluginConfig.Plugin.Kafka.NAMESPACE + "-" + topic; + topics.add(topicName); + return topicName; + } + + public void addListener(KafkaConnectionStatusListener listener) { + if (!listeners.contains(listener)) { + listeners.add(listener); + } + } + + @Override + public void onComplete() { + } + + @Override + public void run() { Properties properties = new Properties(); properties.setProperty( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaReporterPluginConfig.Plugin.Kafka.BOOTSTRAP_SERVERS); - KafkaReporterPluginConfig.Plugin.Kafka.PRODUCER_CONFIG.forEach((k, v) -> properties.setProperty(k, v)); + KafkaReporterPluginConfig.Plugin.Kafka.PRODUCER_CONFIG.forEach(properties::setProperty); AdminClient adminClient = AdminClient.create(properties); - DescribeTopicsResult topicsResult = adminClient.describeTopics(Arrays.asList( - KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_MANAGEMENT, - KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_METRICS, - KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_PROFILING, - KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_SEGMENT, - KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_METER - )); + DescribeTopicsResult topicsResult = adminClient.describeTopics(topics); Set topics = topicsResult.values().entrySet().stream() .map(entry -> { try { @@ -79,24 +119,24 @@ public class KafkaProducerManager implements BootService, Runnable { .filter(Objects::nonNull) .collect(Collectors.toSet()); if (!topics.isEmpty()) { - throw new Exception("These topics" + topics + " don't exist."); + LOGGER.warn("kafka topics {} is not exist, connect to kafka cluster abort", topics); + return; } - producer = new KafkaProducer<>(properties, new StringSerializer(), new BytesSerializer()); + try { + producer = new KafkaProducer<>(properties, new StringSerializer(), new BytesSerializer()); + } catch (Exception e) { + LOGGER.error(e, "connect to kafka cluster '{}' failed", KafkaReporterPluginConfig.Plugin.Kafka.BOOTSTRAP_SERVERS); + } + //notify listeners to send data if no exception been throw + notifyListeners(KafkaConnectionStatus.CONNECTED); + bootProducerFuture.cancel(true); } - @Override - public void boot() { - - } - - @Override - public void onComplete() { - } - - @Override - public void run() { - + private void notifyListeners(KafkaConnectionStatus status) { + for (KafkaConnectionStatusListener listener : listeners) { + listener.onStatusChanged(status); + } } /** @@ -106,6 +146,16 @@ public class KafkaProducerManager implements BootService, Runnable { return producer; } + /** + * make kafka producer init later but before {@link GRPCChannelManager} + * + * @return priority value + */ + @Override + public int priority() { + return ServiceManager.INSTANCE.findService(GRPCChannelManager.class).priority() - 1; + } + @Override public void shutdown() { producer.flush(); diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProfileSnapshotSender.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProfileSnapshotSender.java index 14ff6ffc1..749fe79ef 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProfileSnapshotSender.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProfileSnapshotSender.java @@ -19,6 +19,7 @@ package org.apache.skywalking.apm.agent.core.kafka; import java.util.List; + import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.utils.Bytes; @@ -35,7 +36,7 @@ import org.apache.skywalking.apm.network.language.profile.v3.ThreadSnapshot; * to report the tracing profile snapshot data by Kafka Producer. */ @OverrideImplementor(ProfileSnapshotSender.class) -public class KafkaProfileSnapshotSender extends ProfileSnapshotSender { +public class KafkaProfileSnapshotSender extends ProfileSnapshotSender implements KafkaConnectionStatusListener { private static final ILog LOGGER = LogManager.getLogger(ProfileSnapshotSender.class); private String topic; @@ -43,16 +44,20 @@ public class KafkaProfileSnapshotSender extends ProfileSnapshotSender { @Override public void prepare() { - topic = KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_PROFILING; + KafkaProducerManager producerManager = ServiceManager.INSTANCE.findService(KafkaProducerManager.class); + producerManager.addListener(this); + topic = producerManager.formatTopicNameThenRegister(KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_PROFILING); } @Override public void boot() { - producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); } @Override public void send(final List buffer) { + if (producer == null) { + return; + } for (TracingThreadSnapshot snapshot : buffer) { final ThreadSnapshot object = snapshot.transform(); if (LOGGER.isDebugEnable()) { @@ -69,4 +74,10 @@ public class KafkaProfileSnapshotSender extends ProfileSnapshotSender { } } + @Override + public void onStatusChanged(KafkaConnectionStatus status) { + if (status == KafkaConnectionStatus.CONNECTED) { + producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); + } + } } diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java index e401bde35..5b1d94e0c 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java @@ -42,6 +42,8 @@ public class KafkaReporterPluginConfig { public static String TOPIC_METER = "skywalking-meters"; + public static String NAMESPACE = ""; + public static Map PRODUCER_CONFIG = new HashMap<>(); /** diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaServiceManagementServiceClient.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaServiceManagementServiceClient.java index 5b66b5de9..0d0b48b46 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaServiceManagementServiceClient.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaServiceManagementServiceClient.java @@ -47,7 +47,7 @@ import org.apache.skywalking.apm.util.StringUtil; * A service management data(Instance registering properties and Instance pinging) reporter. */ @OverrideImplementor(ServiceManagementClient.class) -public class KafkaServiceManagementServiceClient implements BootService, Runnable { +public class KafkaServiceManagementServiceClient implements BootService, Runnable, KafkaConnectionStatusListener { private static final ILog LOGGER = LogManager.getLogger(KafkaServiceManagementServiceClient.class); private static List SERVICE_INSTANCE_PROPERTIES; @@ -62,7 +62,9 @@ public class KafkaServiceManagementServiceClient implements BootService, Runnabl @Override public void prepare() { - topic = KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_MANAGEMENT; + KafkaProducerManager producerManager = ServiceManager.INSTANCE.findService(KafkaProducerManager.class); + producerManager.addListener(this); + topic = producerManager.formatTopicNameThenRegister(KafkaReporterPluginConfig.Plugin.Kafka.TOPIC_MANAGEMENT); SERVICE_INSTANCE_PROPERTIES = new ArrayList<>(); for (String key : Config.Agent.INSTANCE_PROPERTIES.keySet()) { @@ -79,8 +81,6 @@ public class KafkaServiceManagementServiceClient implements BootService, Runnabl @Override public void boot() { - producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); - heartbeatFuture = Executors.newSingleThreadScheduledExecutor( new DefaultNamedThreadFactory("ServiceManagementClientKafkaProducer") ).scheduleAtFixedRate(new RunnableWithExceptionProtection( @@ -91,6 +91,9 @@ public class KafkaServiceManagementServiceClient implements BootService, Runnabl @Override public void run() { + if (producer == null) { + return; + } if (Math.abs(sendPropertiesCounter.getAndAdd(1)) % Config.Collector.PROPERTIES_REPORT_PERIOD_FACTOR == 0) { InstanceProperties instance = InstanceProperties.newBuilder() .setService(Config.Agent.SERVICE_NAME) @@ -120,6 +123,13 @@ public class KafkaServiceManagementServiceClient implements BootService, Runnabl } + @Override + public void onStatusChanged(KafkaConnectionStatus status) { + if (status == KafkaConnectionStatus.CONNECTED) { + producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); + } + } + @Override public void shutdown() { heartbeatFuture.cancel(true); diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaTraceSegmentServiceClient.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaTraceSegmentServiceClient.java index c10ebd63c..ac5ae3336 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaTraceSegmentServiceClient.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaTraceSegmentServiceClient.java @@ -44,7 +44,7 @@ import static org.apache.skywalking.apm.agent.core.conf.Config.Buffer.CHANNEL_SI * A tracing segment data reporter. */ @OverrideImplementor(TraceSegmentServiceClient.class) -public class KafkaTraceSegmentServiceClient implements BootService, IConsumer, TracingContextListener { +public class KafkaTraceSegmentServiceClient implements BootService, IConsumer, TracingContextListener, KafkaConnectionStatusListener { private static final ILog LOGGER = LogManager.getLogger(KafkaTraceSegmentServiceClient.class); private String topic; @@ -54,15 +54,15 @@ public class KafkaTraceSegmentServiceClient implements BootService, IConsumer(CHANNEL_SIZE, BUFFER_SIZE, BufferStrategy.IF_POSSIBLE); carrier.consume(this, 1); - - producer = ServiceManager.INSTANCE.findService(KafkaProducerManager.class).getProducer(); } @Override @@ -83,6 +83,9 @@ public class KafkaTraceSegmentServiceClient implements BootService, IConsumer data) { + if (producer == null) { + return; + } data.forEach(traceSegment -> { SegmentObject upstreamSegment = traceSegment.transform(); ProducerRecord record = new ProducerRecord<>( @@ -121,4 +124,10 @@ public class KafkaTraceSegmentServiceClient implements BootService, IConsumer record) { + + } + } +}