Skip to content

Commit a1d69cb

Browse files
authored
Fix agent lifecycle events: Start instance name and Shutdown delivery (#810)
1 parent 1acfda1 commit a1d69cb

6 files changed

Lines changed: 138 additions & 24 deletions

File tree

CHANGES.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ Release Notes.
2626
* Add unified release script (`tools/releasing/release.sh`) with two-step flow: `prepare-vote` and `vote-passed`.
2727
* Fix an issue where `JDBCPluginConfig.Plugin.JDBC.SQL_BODY_MAX_LENGTH` was not honored by clickhouse-0.3.1 and clickhouse-0.3.2.x plugins.
2828
- Add tracing support for vector-store retrieval operations.
29+
* Fix agent lifecycle events: the Start event now carries the service instance name, and the Shutdown event is delivered on graceful JVM exit. `ServiceManager` prepares/starts higher-priority `BootService`s first and shuts them down last (matching `BootService#priority()`), and the shutdown event refreshes its gRPC deadline before sending.
2930

3031
All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/249?closed=1)
3132

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/boot/ServiceManager.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public void boot() {
5050
}
5151

5252
public void shutdown() {
53-
bootedServices.values().stream().sorted(Comparator.comparingInt(BootService::priority).reversed()).forEach(service -> {
53+
bootedServices.values().stream().sorted(Comparator.comparingInt(BootService::priority)).forEach(service -> {
5454
try {
5555
service.shutdown();
5656
} catch (Throwable e) {
@@ -103,7 +103,7 @@ private Map<Class, BootService> loadAllServices() {
103103
}
104104

105105
private void prepare() {
106-
bootedServices.values().stream().sorted(Comparator.comparingInt(BootService::priority)).forEach(service -> {
106+
bootedServices.values().stream().sorted(Comparator.comparingInt(BootService::priority).reversed()).forEach(service -> {
107107
try {
108108
service.prepare();
109109
} catch (Throwable e) {
@@ -113,7 +113,7 @@ private void prepare() {
113113
}
114114

115115
private void startup() {
116-
bootedServices.values().stream().sorted(Comparator.comparingInt(BootService::priority)).forEach(service -> {
116+
bootedServices.values().stream().sorted(Comparator.comparingInt(BootService::priority).reversed()).forEach(service -> {
117117
try {
118118
service.boot();
119119
} catch (Throwable e) {

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/EventReportServiceClient.java

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -50,11 +50,13 @@ public class EventReportServiceClient implements BootService, GRPCChannelListene
5050

5151
private final AtomicBoolean reported = new AtomicBoolean();
5252

53+
private volatile boolean bootCompleted;
54+
5355
private Event.Builder startingEvent;
5456

55-
private EventServiceGrpc.EventServiceStub eventServiceStub;
57+
private volatile EventServiceGrpc.EventServiceStub eventServiceStub;
5658

57-
private GRPCChannelStatus status;
59+
private volatile GRPCChannelStatus status = GRPCChannelStatus.DISCONNECT;
5860

5961
@Override
6062
public void prepare() throws Throwable {
@@ -91,6 +93,7 @@ public void boot() throws Throwable {
9193
@Override
9294
public void onComplete() throws Throwable {
9395
startingEvent.setEndTime(System.currentTimeMillis());
96+
bootCompleted = true;
9497

9598
reportStartingEvent();
9699
}
@@ -117,7 +120,8 @@ public void shutdown() throws Throwable {
117120
)
118121
.setLayer(EVENT_LAYER_NAME);
119122

120-
final StreamObserver<Event> collector = eventServiceStub.collect(new StreamObserver<Commands>() {
123+
final EventServiceGrpc.EventServiceStub stub = eventServiceStub.withDeadlineAfter(GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS);
124+
final StreamObserver<Event> collector = stub.collect(new StreamObserver<Commands>() {
121125
@Override
122126
public void onNext(final Commands commands) {
123127
ServiceManager.INSTANCE.findService(CommandService.class).receiveCommand(commands);
@@ -143,25 +147,27 @@ public void onCompleted() {
143147

144148
@Override
145149
public void statusChanged(final GRPCChannelStatus status) {
150+
if (CONNECTED.equals(status)) {
151+
final Channel channel = ServiceManager.INSTANCE.findService(GRPCChannelManager.class).getChannel();
152+
eventServiceStub = EventServiceGrpc.newStub(channel);
153+
}
146154
this.status = status;
147155

148-
if (!CONNECTED.equals(status)) {
149-
return;
156+
if (CONNECTED.equals(status)) {
157+
reportStartingEvent();
150158
}
151-
152-
final Channel channel = ServiceManager.INSTANCE.findService(GRPCChannelManager.class).getChannel();
153-
eventServiceStub = EventServiceGrpc.newStub(channel);
154-
eventServiceStub = eventServiceStub.withDeadlineAfter(GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS);
155-
156-
reportStartingEvent();
157159
}
158160

159161
private void reportStartingEvent() {
160-
if (reported.compareAndSet(false, true)) {
162+
if (!bootCompleted || !CONNECTED.equals(status)) {
163+
return;
164+
}
165+
if (!reported.compareAndSet(false, true)) {
161166
return;
162167
}
163168

164-
final StreamObserver<Event> collector = eventServiceStub.collect(new StreamObserver<Commands>() {
169+
final EventServiceGrpc.EventServiceStub stub = eventServiceStub.withDeadlineAfter(GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS);
170+
final StreamObserver<Event> collector = stub.collect(new StreamObserver<Commands>() {
165171
@Override
166172
public void onNext(final Commands commands) {
167173
ServiceManager.INSTANCE.findService(CommandService.class).receiveCommand(commands);
Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.agent.core.boot;
20+
21+
import java.lang.reflect.Field;
22+
import java.lang.reflect.Method;
23+
import java.util.ArrayList;
24+
import java.util.Arrays;
25+
import java.util.LinkedHashMap;
26+
import java.util.List;
27+
import java.util.Map;
28+
import org.junit.After;
29+
import org.junit.Test;
30+
31+
import static org.hamcrest.CoreMatchers.is;
32+
import static org.hamcrest.MatcherAssert.assertThat;
33+
34+
public class ServiceManagerOrderingTest {
35+
36+
private final List<String> prepareOrder = new ArrayList<>();
37+
private final List<String> bootOrder = new ArrayList<>();
38+
private final List<String> shutdownOrder = new ArrayList<>();
39+
40+
private class RecordingService implements BootService {
41+
private final String name;
42+
private final int priority;
43+
44+
private RecordingService(String name, int priority) {
45+
this.name = name;
46+
this.priority = priority;
47+
}
48+
49+
@Override
50+
public void prepare() {
51+
prepareOrder.add(name);
52+
}
53+
54+
@Override
55+
public void boot() {
56+
bootOrder.add(name);
57+
}
58+
59+
@Override
60+
public void onComplete() {
61+
}
62+
63+
@Override
64+
public void shutdown() {
65+
shutdownOrder.add(name);
66+
}
67+
68+
@Override
69+
public int priority() {
70+
return priority;
71+
}
72+
}
73+
74+
@After
75+
public void tearDown() throws Exception {
76+
setBootedServices(new LinkedHashMap<>());
77+
}
78+
79+
@Test
80+
public void higherPriorityBootsFirstAndShutsDownLast() throws Exception {
81+
Map<Class, BootService> services = new LinkedHashMap<>();
82+
services.put(Integer.class, new RecordingService("low", 0));
83+
services.put(Long.class, new RecordingService("high", Integer.MAX_VALUE));
84+
services.put(Short.class, new RecordingService("mid", 100));
85+
setBootedServices(services);
86+
87+
invoke("prepare");
88+
invoke("startup");
89+
ServiceManager.INSTANCE.shutdown();
90+
91+
assertThat(prepareOrder, is(Arrays.asList("high", "mid", "low")));
92+
assertThat(bootOrder, is(Arrays.asList("high", "mid", "low")));
93+
assertThat(shutdownOrder, is(Arrays.asList("low", "mid", "high")));
94+
}
95+
96+
private void setBootedServices(Map<Class, BootService> services) throws Exception {
97+
Field field = ServiceManager.class.getDeclaredField("bootedServices");
98+
field.setAccessible(true);
99+
field.set(ServiceManager.INSTANCE, services);
100+
}
101+
102+
private void invoke(String method) throws Exception {
103+
Method m = ServiceManager.class.getDeclaredMethod(method);
104+
m.setAccessible(true);
105+
m.invoke(ServiceManager.INSTANCE);
106+
}
107+
}

apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -44,12 +44,10 @@
4444
import org.apache.skywalking.apm.agent.core.boot.BootService;
4545
import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor;
4646
import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory;
47-
import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
4847
import org.apache.skywalking.apm.agent.core.kafka.KafkaReporterPluginConfig.Plugin.Kafka;
4948
import org.apache.skywalking.apm.agent.core.logging.api.ILog;
5049
import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
5150
import org.apache.skywalking.apm.agent.core.plugin.loader.AgentClassLoader;
52-
import org.apache.skywalking.apm.agent.core.remote.GRPCChannelManager;
5351
import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
5452
import org.apache.skywalking.apm.util.StringUtil;
5553

@@ -183,14 +181,10 @@ public final KafkaProducer<String, Bytes> getProducer() {
183181
return producer;
184182
}
185183

186-
/**
187-
* make kafka producer init later but before {@link GRPCChannelManager}
188-
*
189-
* @return priority value
190-
*/
184+
// Higher than the Kafka reporters sharing this producer, so the producer closes only after they stop.
191185
@Override
192186
public int priority() {
193-
return ServiceManager.INSTANCE.findService(GRPCChannelManager.class).priority() - 1;
187+
return 1;
194188
}
195189

196190
@Override

apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/test/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManagerTest.java

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.skywalking.apm.agent.core.kafka;
2020

2121
import static org.junit.Assert.assertEquals;
22+
import static org.junit.Assert.assertTrue;
2223
import java.lang.reflect.Method;
2324
import java.util.Properties;
2425
import java.util.concurrent.atomic.AtomicInteger;
@@ -46,6 +47,11 @@ public void testAddListener() throws Exception {
4647
assertEquals(counter.get(), times);
4748
}
4849

50+
@Test
51+
public void outranksKafkaReportersSoProducerClosesLast() {
52+
assertTrue(new KafkaProducerManager().priority() > new KafkaTraceSegmentServiceClient().priority());
53+
}
54+
4955
@Test
5056
public void testFormatTopicNameThenRegister() {
5157
KafkaProducerManager kafkaProducerManager = new KafkaProducerManager();

0 commit comments

Comments
 (0)