Skip to content

Commit fb804c6

Browse files
bhardwaj-priyanshubhardwaj-priyanshu
committed
Consolidate Metrics Query and Metrics Proccessor Services
1 parent ea42fd0 commit fb804c6

File tree

7 files changed

+146
-17
lines changed

7 files changed

+146
-17
lines changed

‎cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/MonitorHandlerModule.java

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@
5151
import io.cdap.cdap.internal.app.services.AppFabricServer;
5252
import io.cdap.cdap.logging.run.LogSaverServiceManager;
5353
import io.cdap.cdap.messaging.distributed.MessagingServiceManager;
54-
import io.cdap.cdap.metrics.runtime.MetricsProcessorStatusServiceManager;
5554
import io.cdap.cdap.metrics.runtime.MetricsServiceManager;
5655
import io.cdap.http.HttpHandler;
5756
import java.util.Map;
@@ -115,9 +114,6 @@ private void addNonHadoopBindings(Binder binder) {
115114
.toProvider(new NonHadoopMasterServiceManagerProvider(LogSaverServiceManager.class));
116115
mapBinder.addBinding(Constants.Service.TRANSACTION)
117116
.toProvider(new NonHadoopMasterServiceManagerProvider(TransactionServiceManager.class));
118-
mapBinder.addBinding(Constants.Service.METRICS_PROCESSOR)
119-
.toProvider(
120-
new NonHadoopMasterServiceManagerProvider(MetricsProcessorStatusServiceManager.class));
121117
mapBinder.addBinding(Constants.Service.METRICS)
122118
.toProvider(new NonHadoopMasterServiceManagerProvider(MetricsServiceManager.class));
123119
mapBinder.addBinding(Constants.Service.APP_FABRIC_HTTP)
@@ -152,11 +148,10 @@ private void addHadoopBindings(Binder binder) {
152148
MasterServiceManager.class);
153149
mapBinder.addBinding(Constants.Service.LOGSAVER).to(LogSaverServiceManager.class);
154150
mapBinder.addBinding(Constants.Service.TRANSACTION).to(TransactionServiceManager.class);
155-
mapBinder.addBinding(Constants.Service.METRICS_PROCESSOR)
156-
.to(MetricsProcessorStatusServiceManager.class);
157151
mapBinder.addBinding(Constants.Service.METRICS).to(MetricsServiceManager.class);
158152
mapBinder.addBinding(Constants.Service.APP_FABRIC_HTTP).to(AppFabricServiceManager.class);
159-
mapBinder.addBinding(Constants.Service.APP_FABRIC_PROCESSOR).to(AppFabricProcessorManager.class);
153+
mapBinder.addBinding(Constants.Service.APP_FABRIC_PROCESSOR)
154+
.to(AppFabricProcessorManager.class);
160155
mapBinder.addBinding(Constants.Service.DATASET_EXECUTOR)
161156
.to(DatasetExecutorServiceManager.class);
162157
mapBinder.addBinding(Constants.Service.METADATA_SERVICE).to(MetadataServiceManager.class);

‎cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/services/http/handlers/MonitorHandlerTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ public void testSystemServices() throws Exception {
6666
List<SystemServiceMeta> actual = GSON.fromJson(new String(ByteStreams.toByteArray(urlConn.getInputStream()),
6767
Charsets.UTF_8), token);
6868

69-
Assert.assertEquals(10, actual.size());
69+
Assert.assertEquals(9, actual.size());
7070
urlConn.disconnect();
7171
}
7272

@@ -81,7 +81,7 @@ public void testSystemServicesStatus() throws Exception {
8181

8282
Map<String, String> result = GSON.fromJson(new String(ByteStreams.toByteArray(urlConn.getInputStream()),
8383
Charsets.UTF_8), token);
84-
Assert.assertEquals(10, result.size());
84+
Assert.assertEquals(9, result.size());
8585
urlConn.disconnect();
8686
Assert.assertEquals("OK", result.get(Constants.Service.APP_FABRIC_HTTP));
8787
}

‎cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -129,7 +129,7 @@ public static final class Service {
129129
public static final String LOG_QUERY = "log.query";
130130
public static final String GATEWAY = "gateway";
131131
public static final String MASTER_SERVICES = "master.services";
132-
public static final String METRICS_PROCESSOR = "metrics.processor";
132+
public static final String METRICS_PROCESSOR = "metrics";
133133
public static final String DATASET_MANAGER = "dataset.service";
134134
public static final String DATASET_EXECUTOR = "dataset.executor";
135135
public static final String EXTERNAL_AUTHENTICATION = "external.authentication";

‎cdap-gateway/src/main/java/io/cdap/cdap/gateway/router/RouterPathLookup.java

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -225,8 +225,6 @@ private RouteDestination getV3RoutingService(String[] uriParts, AllowedMethod re
225225
return LOG_SAVER;
226226
case Constants.Service.TRANSACTION:
227227
return TRANSACTION;
228-
case Constants.Service.METRICS_PROCESSOR:
229-
return METRICS_PROCESSOR;
230228
case Constants.Service.METRICS:
231229
return METRICS;
232230
case Constants.Service.APP_FABRIC_HTTP:

‎cdap-master/src/main/java/io/cdap/cdap/master/environment/k8s/MetricsServiceMain.java

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import com.google.inject.Injector;
2222
import com.google.inject.Key;
2323
import com.google.inject.Module;
24+
import com.google.inject.PrivateModule;
25+
import com.google.inject.Scopes;
2426
import com.google.inject.assistedinject.FactoryModuleBuilder;
2527
import io.cdap.cdap.api.metrics.MetricsCollectionService;
2628
import io.cdap.cdap.api.metrics.MetricsContext;
@@ -39,9 +41,8 @@
3941
import io.cdap.cdap.metrics.guice.MetricsStoreModule;
4042
import io.cdap.cdap.metrics.process.MessagingMetricsProcessorServiceFactory;
4143
import io.cdap.cdap.metrics.process.MetricsAdminSubscriberService;
42-
import io.cdap.cdap.metrics.process.MetricsProcessorStatusService;
4344
import io.cdap.cdap.metrics.process.loader.MetricsWriterModule;
44-
import io.cdap.cdap.metrics.query.MetricsQueryService;
45+
import io.cdap.cdap.metrics.service.MetricsService;
4546
import io.cdap.cdap.metrics.store.MetricsCleanUpService;
4647
import io.cdap.cdap.proto.id.NamespaceId;
4748
import io.cdap.cdap.security.authorization.AuthorizationEnforcementModule;
@@ -78,7 +79,14 @@ protected List<Module> getServiceModules(MasterEnvironment masterEnv,
7879
new MetricsProcessorStatusServiceModule(),
7980
new MetricsHandlerModule(),
8081
new DFSLocationModule(),
81-
new MetricsWriterModule()
82+
new MetricsWriterModule(),
83+
new PrivateModule() {
84+
@Override
85+
protected void configure() {
86+
bind(MetricsService.class).in(Scopes.SINGLETON);
87+
expose(MetricsService.class);
88+
}
89+
}
8290
);
8391
}
8492

@@ -98,8 +106,7 @@ protected void addServices(Injector injector, List<? super Service> services,
98106

99107
services.add(injector.getInstance(MessagingMetricsProcessorServiceFactory.class)
100108
.create(topicNumbers, metricsContext, 0));
101-
services.add(injector.getInstance(MetricsProcessorStatusService.class));
102-
services.add(injector.getInstance(MetricsQueryService.class));
109+
services.add(injector.getInstance(MetricsService.class));
103110
services.add(injector.getInstance(MetricsAdminSubscriberService.class));
104111
services.add(injector.getInstance(MetricsCleanUpService.class));
105112
Binding<ZKClientService> zkBinding = injector.getExistingBinding(

‎cdap-watchdog/src/main/java/io/cdap/cdap/metrics/query/MetricsQueryService.java

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ public class MetricsQueryService extends AbstractIdleService {
4545
private static final Logger LOG = LoggerFactory.getLogger(MetricsQueryService.class);
4646

4747
private final NettyHttpService httpService;
48+
private final Set<HttpHandler> handlers;
4849
private final DiscoveryService discoveryService;
4950
private Cancellable cancelDiscovery;
5051

@@ -53,6 +54,7 @@ public MetricsQueryService(CConfiguration cConf, SConfiguration sConf,
5354
@Named(Constants.Service.METRICS) Set<HttpHandler> handlers,
5455
DiscoveryService discoveryService,
5556
CommonNettyHttpServiceFactory commonNettyHttpServiceFactory) {
57+
this.handlers = handlers;
5658
// netty http server config
5759
String address = cConf.get(Constants.Metrics.ADDRESS);
5860
int backlogcnxs = cConf.getInt(Constants.Metrics.BACKLOG_CONNECTIONS, 20000);
@@ -109,4 +111,11 @@ protected void shutDown() throws Exception {
109111
cancelDiscovery.cancel();
110112
httpService.stop();
111113
}
114+
115+
/**
116+
* Returns the set of HTTP handlers associated.
117+
*/
118+
public Set<HttpHandler> getHandlers() {
119+
return handlers;
120+
}
112121
}

‎cdap-watchdog/src/main/java/io/cdap/cdap/metrics/service/MetricsService.java

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
/*
2+
* Copyright © 2025 Cask Data, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package io.cdap.cdap.metrics.service;
18+
19+
import com.google.common.util.concurrent.AbstractIdleService;
20+
import com.google.inject.Inject;
21+
import io.cdap.cdap.common.conf.CConfiguration;
22+
import io.cdap.cdap.common.conf.Constants;
23+
import io.cdap.cdap.common.conf.SConfiguration;
24+
import io.cdap.cdap.common.discovery.ResolvingDiscoverable;
25+
import io.cdap.cdap.common.discovery.URIScheme;
26+
import io.cdap.cdap.common.http.CommonNettyHttpServiceFactory;
27+
import io.cdap.cdap.common.id.Id;
28+
import io.cdap.cdap.common.logging.LoggingContextAccessor;
29+
import io.cdap.cdap.common.logging.ServiceLoggingContext;
30+
import io.cdap.cdap.common.security.HttpsEnabler;
31+
import io.cdap.cdap.metrics.query.MetricsQueryService;
32+
import io.cdap.http.HttpHandler;
33+
import io.cdap.http.NettyHttpService;
34+
import java.util.Set;
35+
import org.apache.twill.common.Cancellable;
36+
import org.apache.twill.discovery.DiscoveryService;
37+
import org.slf4j.Logger;
38+
import org.slf4j.LoggerFactory;
39+
40+
/**
41+
* Metrics implemented using the common http netty framework.
42+
*/
43+
public class MetricsService extends AbstractIdleService {
44+
45+
private static final Logger LOG = LoggerFactory.getLogger(MetricsService.class);
46+
47+
private final NettyHttpService httpService;
48+
private final DiscoveryService discoveryService;
49+
private final MetricsQueryService metricsQueryService;
50+
private Cancellable cancelDiscovery;
51+
52+
@Inject
53+
public MetricsService(CConfiguration cConf, SConfiguration sConf,
54+
MetricsQueryService metricsQueryService,
55+
DiscoveryService discoveryService,
56+
CommonNettyHttpServiceFactory commonNettyHttpServiceFactory) {
57+
this.metricsQueryService = metricsQueryService;
58+
// netty http server config
59+
String address = cConf.get(Constants.Metrics.ADDRESS);
60+
int backlogcnxs = cConf.getInt(Constants.Metrics.BACKLOG_CONNECTIONS, 20000);
61+
int execthreads = cConf.getInt(Constants.Metrics.EXEC_THREADS, 20);
62+
int bossthreads = cConf.getInt(Constants.Metrics.BOSS_THREADS, 1);
63+
int workerthreads = cConf.getInt(Constants.Metrics.WORKER_THREADS, 10);
64+
Set<HttpHandler> handlers = metricsQueryService.getHandlers();
65+
66+
NettyHttpService.Builder builder = commonNettyHttpServiceFactory.builder(
67+
Constants.Service.METRICS)
68+
.setHttpHandlers(handlers)
69+
.setHost(address)
70+
.setPort(cConf.getInt(Constants.Metrics.PORT))
71+
.setConnectionBacklog(backlogcnxs)
72+
.setExecThreadPoolSize(execthreads)
73+
.setBossThreadPoolSize(bossthreads)
74+
.setWorkerThreadPoolSize(workerthreads);
75+
76+
if (cConf.getBoolean(Constants.Security.SSL.INTERNAL_ENABLED)) {
77+
new HttpsEnabler().configureKeyStore(cConf, sConf).enable(builder);
78+
}
79+
80+
this.httpService = builder.build();
81+
this.discoveryService = discoveryService;
82+
83+
LOG.info("Configuring MetricsService "
84+
+ ", address: " + address
85+
+ ", backlog connections: " + backlogcnxs
86+
+ ", execthreads: " + execthreads
87+
+ ", bossthreads: " + bossthreads
88+
+ ", workerthreads: " + workerthreads);
89+
}
90+
91+
@Override
92+
protected void startUp() throws Exception {
93+
LoggingContextAccessor.setLoggingContext(new ServiceLoggingContext(Id.Namespace.SYSTEM.getId(),
94+
Constants.Logging.COMPONENT_NAME,
95+
Constants.Service.METRICS));
96+
97+
LOG.info("Starting Metrics Service...");
98+
httpService.start();
99+
LOG.info("Started Metrics HTTP Service...");
100+
// Register the service
101+
cancelDiscovery = discoveryService.register(
102+
ResolvingDiscoverable.of(
103+
URIScheme.createDiscoverable(Constants.Service.METRICS, httpService)));
104+
LOG.info("Metrics Service started successfully on {}", httpService.getBindAddress());
105+
}
106+
107+
@Override
108+
protected void shutDown() throws Exception {
109+
LOG.info("Stopping Metrics Service...");
110+
111+
// Unregister the service
112+
try {
113+
if (cancelDiscovery != null) {
114+
cancelDiscovery.cancel();
115+
}
116+
} finally {
117+
httpService.stop();
118+
}
119+
}
120+
}

0 commit comments

Comments
 (0)