Skip to content
This repository was archived by the owner on Jan 8, 2020. It is now read-only.

Commit fd43c85

Browse files
mstumpmstump
committed
added support for Kafka in metrics3 and associated sample config
1 parent 479358a commit fd43c85

12 files changed

Lines changed: 371 additions & 6 deletions

File tree

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,3 +10,4 @@ target/
1010
pom.xml.releaseBackup
1111
release.properties
1212
TODO
13+
.DS_Store

‎pom.xml‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,6 @@
115115
<version>3.1</version>
116116
</dependency>
117117
</dependencies>
118-
119118
<repositories>
120119
<!-- re-specify central repo to ensure it's the default -->
121120
<repository>
@@ -127,6 +126,11 @@
127126
<enabled>false</enabled>
128127
</snapshots>
129128
</repository>
129+
<!-- kafka reporter -->
130+
<repository>
131+
<id>jitpack.io</id>
132+
<url>https://jitpack.io</url>
133+
</repository>
130134
<!-- riemann reporter -->
131135
<repository>
132136
<id>clojars.org</id>

‎reporter-config-base/src/main/java/com/addthis/metrics/reporter/config/AbstractHostPortReporterConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,7 +172,7 @@ private String sanitizeName(String name) {
172172
return name.replaceAll("[^a-zA-Z0-9_-]", "_");
173173
}
174174

175-
String resolvePrefix(String prefixTemplate) {
175+
public String resolvePrefix(String prefixTemplate) {
176176
Map<String, String> valueMap = new HashMap<String, String>();
177177
if (localhost != null) {
178178
String hostname = localhost.getHostName();

‎reporter-config-base/src/main/java/com/addthis/metrics/reporter/config/AbstractKafkaReporterConfig.java‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
/*
2+
* Licensed under the Apache License, Version 2.0 (the "License");
3+
* you may not use this file except in compliance with the License.
4+
* You may obtain a copy of the License at
5+
*
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* Unless required by applicable law or agreed to in writing, software
9+
* distributed under the License is distributed on an "AS IS" BASIS,
10+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
* See the License for the specific language governing permissions and
12+
* limitations under the License.
13+
*/
14+
15+
package com.addthis.metrics.reporter.config;
16+
17+
import javax.validation.constraints.NotNull;
18+
19+
import java.util.HashMap;
20+
import java.util.List;
21+
import java.util.Map;
22+
23+
public abstract class AbstractKafkaReporterConfig extends AbstractHostPortReporterConfig {
24+
25+
protected String name = "kafka";
26+
27+
protected String hostname = null;
28+
29+
protected String ip = null;
30+
31+
@NotNull
32+
protected String serializer = "kafka.serializer.StringEncoder";
33+
34+
@NotNull
35+
protected String partitioner = "kafka.producer.DefaultPartitioner";
36+
37+
@NotNull
38+
protected String requiredAcks = "1";
39+
40+
@NotNull
41+
protected String topic;
42+
43+
@NotNull
44+
private Map<String, String> labels;
45+
46+
private Map<String, String> resolvedLabels;
47+
48+
public String getName() {
49+
return name;
50+
}
51+
52+
public void setName(String name) {
53+
this.name = name;
54+
}
55+
56+
public String getTopic() {
57+
return this.topic;
58+
}
59+
60+
public void setTopic(String topic) {
61+
this.topic = topic;
62+
}
63+
64+
public String getHostname() {
65+
return this.hostname;
66+
}
67+
68+
public void setHostname(String hostname) {
69+
if (hostname != null) {
70+
this.hostname = resolvePrefix(hostname);
71+
}
72+
this.hostname = null;
73+
}
74+
75+
public String getIp() {
76+
return this.ip;
77+
}
78+
79+
public void setIp(String ip) {
80+
if (ip != null) {
81+
this.ip = resolvePrefix(ip);
82+
}
83+
this.ip = null;
84+
}
85+
86+
public String getSerializer() {
87+
return this.serializer;
88+
}
89+
90+
public void setSerializer(String serializer) {
91+
this.serializer = serializer;
92+
}
93+
94+
public String getPartitioner() {
95+
return this.partitioner;
96+
}
97+
98+
public void setPartitioner(String partitioner) {
99+
this.partitioner = partitioner;
100+
}
101+
102+
public String getRequiredAcks() {
103+
return this.requiredAcks;
104+
}
105+
106+
public void setRequiredAcks(String requiredAcks) {
107+
this.requiredAcks = requiredAcks;
108+
}
109+
110+
public void setLabels(Map<String, String> labels) {
111+
this.labels = labels;
112+
this.resolvedLabels = new HashMap<String, String>(labels.size());
113+
for (Map.Entry<String, String> entry : labels.entrySet())
114+
{
115+
this.resolvedLabels.put(entry.getKey(), resolvePrefix(entry.getValue()));
116+
}
117+
}
118+
119+
public Map<String, String> getResolvedLabels() {
120+
return resolvedLabels;
121+
}
122+
123+
@Override
124+
public List<HostPort> getFullHostList()
125+
{
126+
return getHostListAndStringList();
127+
}
128+
}

‎reporter-config-base/src/main/java/com/addthis/metrics/reporter/config/HostPort.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,9 @@ public void setPort(int port)
5454
{
5555
this.port = port;
5656
}
57-
}
58-
5957

58+
public String toString()
59+
{
60+
return String.format("%s:%d", this.host, this.port);
61+
}
62+
}

‎reporter-config-base/src/test/resources/sample/kafka.yaml‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
kafka:
2+
-
3+
requiredAcks: '1'
4+
topic: 'cassandra'
5+
period: 10
6+
timeunit: 'SECONDS'
7+
hostname: '${host.fqdn}'
8+
ip: '${host.name}'
9+
hosts:
10+
- host: 'localhost'
11+
port: 9092
12+
13+
labels:
14+
'TEST LABEL' : 'TEST VALUE'
15+
16+
predicate:
17+
color: 'white'
18+
useQualifiedName: true
19+
patterns:
20+
- '^org\.apache\.cassandra\.metrics\.Cache.+'
21+
- '^org\.apache\.cassandra\.metrics\.ClientRequest.+'
22+
- '^org\.apache\.cassandra\.metrics\.CommitLog.+'
23+
- '^org\.apache\.cassandra\.metrics\.Compaction.+'
24+
- '^org\.apache\.cassandra\.metrics\.DroppedMetrics.+'
25+
- '^org\.apache\.cassandra\.metrics\.ReadRepair.+'
26+
- '^org\.apache\.cassandra\.metrics\.Storage.+'
27+
- '^org\.apache\.cassandra\.metrics\.ThreadPools.+'
28+
- '^org\.apache\.cassandra\.metrics\.CQL.+'
29+
- '^org\.apache\.cassandra\.metrics\.Client.+'
30+
- '^org\.apache\.cassandra\.metrics\.Table\.[a-zA-Z]+\.all'

‎reporter-config3/pom.xml‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,25 @@
2323
<artifactId>reporter-config3</artifactId>
2424
<name>metrics reporter config 3.x</name>
2525
<description />
26-
2726
<dependencies>
2827
<dependency>
2928
<groupId>com.addthis.metrics</groupId>
3029
<artifactId>reporter-config-base</artifactId>
3130
<version>3.0.3</version>
3231
</dependency>
32+
<dependency>
33+
<groupId>org.apache.kafka</groupId>
34+
<artifactId>kafka_2.10</artifactId>
35+
<version>0.8.2.2</version>
36+
<exclusions>
37+
<exclusion>
38+
<groupId>org.apache.zookeeper</groupId>
39+
<artifactId>zookeeper</artifactId>
40+
</exclusion>
41+
</exclusions>
42+
<scope>test</scope>
43+
<optional>true</optional>
44+
</dependency>
3345
<dependency>
3446
<groupId>io.dropwizard.metrics</groupId>
3547
<artifactId>metrics-core</artifactId>
@@ -93,6 +105,14 @@
93105
<artifactId>simpleclient_servlet</artifactId>
94106
<optional>true</optional>
95107
</dependency>
108+
<!-- kafka -->
109+
<dependency>
110+
<groupId>com.github.mstump</groupId>
111+
<artifactId>metrics-kafka</artifactId>
112+
<version>c804bf1874</version>
113+
<optional>true</optional>
114+
<scope>compile</scope>
115+
</dependency>
96116
</dependencies>
97117

98118
</project>

‎reporter-config3/src/main/java/com/addthis/metrics3/reporter/config/KafkaReporterConfig.java‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
/*
2+
* Licensed under the Apache License, Version 2.0 (the "License");
3+
* you may not use this file except in compliance with the License.
4+
* You may obtain a copy of the License at
5+
*
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* Unless required by applicable law or agreed to in writing, software
9+
* distributed under the License is distributed on an "AS IS" BASIS,
10+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
* See the License for the specific language governing permissions and
12+
* limitations under the License.
13+
*/
14+
15+
package com.addthis.metrics3.reporter.config;
16+
17+
import java.util.List;
18+
import java.util.LinkedList;
19+
import java.util.Properties;
20+
import java.util.StringJoiner;
21+
22+
import com.codahale.metrics.MetricRegistry;
23+
import com.addthis.metrics.reporter.config.HostPort;
24+
import com.addthis.metrics.reporter.config.AbstractKafkaReporterConfig;
25+
26+
import org.slf4j.Logger;
27+
import org.slf4j.LoggerFactory;
28+
29+
import io.github.hengyunabc.metrics.KafkaReporter;
30+
import kafka.producer.ProducerConfig;
31+
32+
33+
public class KafkaReporterConfig extends AbstractKafkaReporterConfig implements MetricsReporterConfigThree
34+
{
35+
private static final Logger log = LoggerFactory.getLogger(KafkaReporterConfig.class);
36+
37+
private MetricRegistry registry;
38+
39+
private KafkaReporter reporter;
40+
41+
private boolean checkClass(String className) {
42+
if (!isClassAvailable(className))
43+
{
44+
log.error("Tried to enable InfluxDBReporter, but class {} was not found", className);
45+
return false;
46+
} else
47+
{
48+
return true;
49+
}
50+
}
51+
52+
@Override
53+
public boolean enable(MetricRegistry registry) {
54+
this.registry = registry;
55+
56+
boolean success = checkClass("com.addthis.metrics.reporter.config.AbstractKafkaReporterConfig");
57+
if (!success)
58+
{
59+
return false;
60+
}
61+
62+
List<HostPort> hosts = getFullHostList();
63+
if (hosts == null || hosts.isEmpty())
64+
{
65+
log.error("No hosts specified, cannot enable KafkaReporter");
66+
return false;
67+
}
68+
69+
log.info("Enabling KafkaReporter to {}", "");
70+
try
71+
{
72+
StringJoiner brokerList = new StringJoiner(",");
73+
for (HostPort host : getFullHostList()) {
74+
brokerList.add(host.toString());
75+
}
76+
77+
Properties props = new Properties();
78+
props.put("metadata.broker.list", brokerList.toString());
79+
props.put("serializer.class", getSerializer());
80+
props.put("partitioner.class", getPartitioner());
81+
props.put("request.required.acks", getRequiredAcks());
82+
ProducerConfig config = new ProducerConfig(props);
83+
84+
reporter = KafkaReporter.forRegistry(registry)
85+
.config(config)
86+
.topic(getTopic())
87+
.hostName(getHostname())
88+
.ip(getIp())
89+
.labels(getResolvedLabels())
90+
.prefix(getResolvedPrefix())
91+
.filter(MetricFilterTransformer.generateFilter(getPredicate()))
92+
.build();
93+
94+
reporter.start(getPeriod(), getRealTimeunit());
95+
}
96+
catch (Exception e)
97+
{
98+
log.error("Failure while Enabling KafkaReporter", e);
99+
return false;
100+
}
101+
return true;
102+
}
103+
104+
@Override
105+
public void report() {
106+
if (reporter != null) {
107+
reporter.report();
108+
}
109+
}
110+
111+
}

0 commit comments

Comments
 (0)