diff --git a/Jenkinsfile-Agent-Test-3 b/Jenkinsfile-Agent-Test-3 index 2b14f2829..64d5846c9 100755 --- a/Jenkinsfile-Agent-Test-3 +++ b/Jenkinsfile-Agent-Test-3 @@ -58,9 +58,9 @@ pipeline { parallel { stage('Group1') { stages { - stage('reserve stages') { + stage('kafka 0.11.0.0-2.3.0 (16)') { steps { - echo "reserve." + sh 'bash test/plugin/run.sh --build_id=${BUILD_ID} kafka-scenario' } } } diff --git a/test/plugin/runner-helper/src/main/resources/docker-compose.template b/test/plugin/runner-helper/src/main/resources/docker-compose.template index d1f53fb1c..c1e1029a8 100644 --- a/test/plugin/runner-helper/src/main/resources/docker-compose.template +++ b/test/plugin/runner-helper/src/main/resources/docker-compose.template @@ -57,9 +57,9 @@ services: - ${volume} - <#if service.environments??> + <#if service.environment??> environment: - <#list service.environments as environment> + <#list service.environment as environment> - ${environment} diff --git a/test/plugin/scenarios/kafka-scenario/bin/startup.sh b/test/plugin/scenarios/kafka-scenario/bin/startup.sh new file mode 100644 index 000000000..11c8849b4 --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/bin/startup.sh @@ -0,0 +1,21 @@ +#!/bin/bash +# +# 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. + +home="$(cd "$(dirname $0)"; pwd)" + +java -Dbootstrap.servers=${BOOTSTRAP_SERVERS} -jar ${agent_opts} "-Dskywalking.agent.service_name=kafka-scenario" ${home}/../libs/kafka-scenario.jar & \ No newline at end of file diff --git a/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml b/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml new file mode 100644 index 000000000..bcd8b492d --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml @@ -0,0 +1,130 @@ +# 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. +registryItems: + applications: + - {kafka-scenario: 2} + instances: + - {kafka-scenario: 1} + operationNames: + - kafka-scenario: [Kafka/test/Producer, Kafka/test/Consumer/testGroup, + /case/healthCheck, /case/kafka-case] + heartbeat: [] +segmentItems: + - applicationCode: kafka-scenario + segmentSize: gt 4 + segments: + - segmentId: not null + spans: + - operationName: /case/healthCheck + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: nq 0 + endTime: nq 0 + componentId: 14 + componentName: '' + isError: false + spanType: Entry + peer: '' + peerId: 0 + tags: + - {key: url, value: 'http://localhost:8080/kafka-scenario/case/healthCheck'} + - {key: http.method, value: HEAD} + - segmentId: not null + spans: + - operationName: Kafka/Producer/Callback + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Unknown + startTime: nq 0 + endTime: nq 0 + componentId: 40 + componentName: '' + isError: false + spanType: Local + peer: '' + peerId: 0 + tags: + - {key: mq.topic, value: test} + refs: + - {parentEndpointId: 0, parentEndpoint: /case/kafka-case, networkAddressId: 0, + entryEndpointId: 0, refType: CrossThread, parentSpanId: 1, parentTraceSegmentId: not null, + parentServiceInstanceId: 1, networkAddress: '', entryEndpoint: /case/kafka-case, + entryServiceInstanceId: 1} + - segmentId: not null + spans: + - operationName: Kafka/test/Producer + operationId: 0 + parentSpanId: 0 + spanId: 1 + spanLayer: MQ + startTime: nq 0 + endTime: nq 0 + componentId: 40 + componentName: '' + isError: false + spanType: Exit + peer: kafka-server:9092 + peerId: 0 + tags: + - {key: mq.broker, value: 'kafka-server:9092'} + - {key: mq.topic, value: test} + - operationName: /case/kafka-case + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: nq 0 + endTime: nq 0 + componentId: 14 + componentName: '' + isError: false + spanType: Entry + peer: '' + peerId: 0 + tags: + - {key: url, value: 'http://localhost:8080/kafka-scenario/case/kafka-case'} + - {key: http.method, value: GET} + - segmentId: not null + spans: + - operationName: Kafka/test/Consumer/testGroup + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: MQ + startTime: nq 0 + endTime: nq 0 + componentId: 41 + componentName: '' + isError: false + spanType: Entry + peer: '' + peerId: 0 + tags: + - {key: mq.broker, value: 'kafka-server:9092'} + - {key: mq.topic, value: test} + refs: + - {parentEndpointId: 0, parentEndpoint: /case/kafka-case, networkAddressId: 0, + entryEndpointId: 0, refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, + parentServiceInstanceId: 1, networkAddress: 'kafka-server:9092', entryEndpoint: /case/kafka-case, + entryServiceInstanceId: 1 + } + + + + diff --git a/test/plugin/scenarios/kafka-scenario/configuration.yml b/test/plugin/scenarios/kafka-scenario/configuration.yml new file mode 100644 index 000000000..40ea5351b --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/configuration.yml @@ -0,0 +1,40 @@ +# 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. + +type: jvm +entryService: http://localhost:8080/kafka-scenario/case/kafka-case +healthCheck: http://localhost:8080/kafka-scenario/case/healthCheck +startScript: ./bin/startup.sh +framework: kafka +environment: + - BOOTSTRAP_SERVERS=kafka-server:9092 +depends_on: + - zookeeper-server + - kafka-server +dependencies: + zookeeper-server: + image: zookeeper:3.4 + hostname: zookeeper-server + kafka-server: + image: bitnami/kafka:2.1.1 + hostname: kafka-server + environment: + - KAFKA_ZOOKEEPER_CONNECT=zookeeper-server:2181 + - KAFKA_BROKER_ID=1 + - ALLOW_PLAINTEXT_LISTENER=yes + - KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 + depends_on: + - zookeeper-server \ No newline at end of file diff --git a/test/plugin/scenarios/kafka-scenario/pom.xml b/test/plugin/scenarios/kafka-scenario/pom.xml new file mode 100644 index 000000000..bf82bcd75 --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/pom.xml @@ -0,0 +1,139 @@ + + + + 4.0.0 + + org.apache.skywalking + kafka-scenario + 5.0.0 + + + UTF-8 + 1.8 + 1.0.0 + spring + + 2.6.2 + 4.3.8.RELEASE + 1.5.2.RELEASE + + + skywalking-kafka-scenario + + + + org.apache.kafka + kafka-clients + ${test.framework.version} + + + + org.springframework.boot + spring-boot-starter + ${spring-boot-version} + + + org.apache.logging.log4j + log4j-api + ${log4j.version} + + + org.apache.logging.log4j + log4j-core + ${log4j.version} + + + org.apache.logging.log4j + log4j-slf4j-impl + ${log4j.version} + + + org.apache.logging.log4j + log4j-jcl + ${log4j.version} + + + org.springframework.boot + spring-boot-starter-tomcat + ${spring-boot-version} + + + org.springframework.boot + spring-boot-starter-web + ${spring-boot-version} + + + + + kafka-scenario + + + org.springframework.boot + spring-boot-maven-plugin + + + + repackage + + + + + + maven-compiler-plugin + + ${compiler.version} + ${compiler.version} + ${project.build.sourceEncoding} + + + + org.apache.maven.plugins + maven-assembly-plugin + + + assemble + package + + single + + + + src/main/assembly/assembly.xml + + ./target/ + + + + + + + + + + spring-snapshots + http://repo.spring.io/snapshot + + + spring-milestones + http://repo.spring.io/milestone + + + \ No newline at end of file diff --git a/test/plugin/scenarios/kafka-scenario/src/main/assembly/assembly.xml b/test/plugin/scenarios/kafka-scenario/src/main/assembly/assembly.xml new file mode 100644 index 000000000..4eca70aab --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/src/main/assembly/assembly.xml @@ -0,0 +1,41 @@ + + + + + zip + + + + + ./bin + 0775 + + + + + + ${project.build.directory}/kafka-scenario.jar + ./libs + 0775 + + + diff --git a/test/plugin/scenarios/kafka-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/kafka/Application.java b/test/plugin/scenarios/kafka-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/kafka/Application.java new file mode 100644 index 000000000..46646d365 --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/kafka/Application.java @@ -0,0 +1,33 @@ +/* + * 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 test.org.apache.skywalking.apm.testcase.kafka; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * @author zhangwei + */ +@SpringBootApplication +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } +} diff --git a/test/plugin/scenarios/kafka-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/kafka/controller/CaseController.java b/test/plugin/scenarios/kafka-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/kafka/controller/CaseController.java new file mode 100644 index 000000000..55b51dd84 --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/kafka/controller/CaseController.java @@ -0,0 +1,193 @@ +/* + * 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 test.org.apache.skywalking.apm.testcase.kafka.controller; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.PropertySource; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.ResponseBody; + +import javax.annotation.PostConstruct; +import java.util.Arrays; +import java.util.Properties; +import java.util.function.Consumer; + +import static java.util.Objects.isNull; + +@Controller +@RequestMapping("/case") +@PropertySource("classpath:application.properties") +public class CaseController { + + private static final Logger logger = LogManager.getLogger(CaseController.class); + + private static final String SUCCESS = "Success"; + + @Value("${bootstrap.servers:127.0.0.1:9092}") + private String bootstrapServers; + + private String topicName; + + private static volatile boolean KAFKA_STATUS = false; + + @PostConstruct + private void setUp() { + topicName = "test"; + new CheckKafkaProducerThread(bootstrapServers).start(); + } + + @RequestMapping("/kafka-case") + @ResponseBody + public String kafkaCase() { + wrapProducer(producer -> { + ProducerRecord record = new ProducerRecord(topicName, "testKey", Integer.toString(1)); + record.headers().add("TEST", "TEST".getBytes()); + producer.send(record, new Callback() { + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + logger.info("send success metadata={}", metadata); + } + }); + }, bootstrapServers); + Thread thread = new ConsumerThread(); + thread.start(); + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + return SUCCESS; + } + + @RequestMapping("/healthCheck") + @ResponseBody + public String healthCheck() { + if (KAFKA_STATUS) { + return SUCCESS; + } + throw new RuntimeException("kafka not ready"); + } + + private static void wrapProducer(Consumer> consFunc, String bootstrapServers) { + Properties producerProperties = new Properties(); + producerProperties.put("bootstrap.servers", bootstrapServers); + producerProperties.put("acks", "all"); + producerProperties.put("retries", 0); + producerProperties.put("batch.size", 16384); + producerProperties.put("linger.ms", 1); + producerProperties.put("buffer.memory", 33554432); + producerProperties.put("auto.create.topics.enable", "true"); + producerProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + producerProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + + Producer producer = new KafkaProducer<>(producerProperties); + try { + consFunc.accept(producer); + } finally { + producer.close(); + } + } + + public static class CheckKafkaProducerThread extends Thread { + + private final String bootstrapServers; + + public CheckKafkaProducerThread(String bootstrapServers) { + setDaemon(true); + this.bootstrapServers = bootstrapServers; + } + + @Override + public void run() { + while (true) { + try { + if (KAFKA_STATUS) { + return; + } + wrapProducer(producer -> { + ProducerRecord record = new ProducerRecord("check", "checkKey", Integer.toString(1)); + record.headers().add("CHECK", "CHECK".getBytes()); + producer.send(record, new Callback() { + @Override + public void onCompletion(RecordMetadata recordMetadata, Exception e) { + if (isNull(e)) { + KAFKA_STATUS = true; + } + } + }); + }, bootstrapServers); + } catch (Exception e) { + logger.error("check " + bootstrapServers + " " + e.getMessage(), e); + } + try { + Thread.sleep(5000); + } catch (InterruptedException e) { + // ignore + } + } + } + } + + public class ConsumerThread extends Thread { + @Override + public void run() { + Properties consumerProperties = new Properties(); + consumerProperties.put("bootstrap.servers", bootstrapServers); + consumerProperties.put("group.id", "testGroup"); + consumerProperties.put("enable.auto.commit", "true"); + consumerProperties.put("auto.commit.interval.ms", "1000"); + consumerProperties.put("auto.offset.reset", "earliest"); + consumerProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + consumerProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + KafkaConsumer consumer = new KafkaConsumer<>(consumerProperties); + consumer.subscribe(Arrays.asList(topicName)); + int i = 0; + while (i++ <= 10) { + try { + Thread.sleep(1 * 1000); + } catch (InterruptedException e) { + } + + ConsumerRecords records = consumer.poll(100); + + if (!records.isEmpty()) { + for (ConsumerRecord record : records) { + logger.info("header: {}", new String(record.headers().headers("TEST").iterator().next().value())); + logger.info("offset = {}, key = {}, value = {}", record.offset(), record.key(), record.value()); + } + break; + } + } + + consumer.close(); + } + } +} + diff --git a/test/plugin/scenarios/kafka-scenario/src/main/resources/application.properties b/test/plugin/scenarios/kafka-scenario/src/main/resources/application.properties new file mode 100644 index 000000000..d7b9a3aea --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/src/main/resources/application.properties @@ -0,0 +1,20 @@ +# +# 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. +# +# + +server.port=8080 +server.contextPath=/kafka-scenario \ No newline at end of file diff --git a/test/plugin/scenarios/kafka-scenario/src/main/resources/log4j2.xml b/test/plugin/scenarios/kafka-scenario/src/main/resources/log4j2.xml new file mode 100644 index 000000000..b5cda5ae8 --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/src/main/resources/log4j2.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + diff --git a/test/plugin/scenarios/kafka-scenario/support-version.list b/test/plugin/scenarios/kafka-scenario/support-version.list new file mode 100644 index 000000000..adc50a13e --- /dev/null +++ b/test/plugin/scenarios/kafka-scenario/support-version.list @@ -0,0 +1,32 @@ +# 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 +# "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. + +0.11.0.0 +0.11.0.1 +0.11.0.2 +0.11.0.3 +1.0.0 +1.0.1 +1.0.2 +1.1.0 +1.1.1 +2.0.0 +2.0.1 +2.1.0 +2.1.1 +2.2.0 +2.2.1 +2.3.0 \ No newline at end of file