Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

update-etl #616

Merged
merged 1 commit into from
Aug 31, 2023
Merged

update-etl #616

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions hera-all/trace-etl/trace-etl-server/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,14 @@
<version>1.0.0-SNAPSHOT</version>
</dependency>


<dependency>
<groupId>com.squareup.okio</groupId>
<artifactId>okio</artifactId>
<version>3.5.0</version>
</dependency>


</dependencies>


Expand Down
38 changes: 31 additions & 7 deletions ...tl/trace-etl-server/src/main/java/com/xiaomi/hera/trace/etl/consumer/ConsumerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,7 @@
import com.xiaomi.hera.tspandata.TSpanData;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.thrift.TDeserializer;
Expand All @@ -19,6 +17,9 @@

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
* @author dingtao
Expand All @@ -44,6 +45,12 @@ public class ConsumerService {
@Autowired
private ClientMessageQueue clientMessageQueue;

@Autowired
private DataCacheService cacheService;

@Autowired
private EnterManager enterManager;

@PostConstruct
public void takeMessage() throws MQClientException {
// Before initializing rocketmq consumer,
Expand All @@ -60,13 +67,19 @@ public void takeMessage() throws MQClientException {
log.info("init consumer end ...");
}

private class TraceEtlMessageListener implements MessageListenerConcurrently {
private class TraceEtlMessageListener implements MessageListenerOrderly {

private AtomicInteger i = new AtomicInteger();

private AtomicLong time = new AtomicLong(System.currentTimeMillis());


@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, final ConsumeOrderlyContext context) {
if (list == null || list.isEmpty()) {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
return ConsumeOrderlyStatus.SUCCESS;
}
enterManager.enter();
for (MessageExt message : list) {
String traceId = "";
try {
Expand All @@ -79,7 +92,18 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeCo
}
clientMessageQueue.enqueue(traceId, message);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

long now = System.currentTimeMillis();

i.addAndGet(list.size());
if ((now - time.get() >= TimeUnit.SECONDS.toMillis(15))) {
i.set(0);
time.set(now);
cacheService.cacheData();
}

enterManager.processEnter();
return ConsumeOrderlyStatus.SUCCESS;
}
}
}
157 changes: 157 additions & 0 deletions ...l/trace-etl-server/src/main/java/com/xiaomi/hera/trace/etl/consumer/DataCacheService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
package com.xiaomi.hera.trace.etl.consumer;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Sets;
import com.xiaomi.youpin.prometheus.client.MetricsManager;
import com.xiaomi.youpin.prometheus.client.Prometheus;
import com.xiaomi.youpin.prometheus.client.multi.MutiMetrics;
import io.prometheus.client.*;
import io.prometheus.client.exporter.common.TextFormat;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.ByteArrayOutputStream;
import java.io.OutputStreamWriter;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
* @author goodjava@qq.com
* @date 2023/8/29 10:02
*/
@Service
@Slf4j
public class DataCacheService {

private CopyOnWriteArrayList<byte[]> cacheData = new CopyOnWriteArrayList<>();

@Resource
private MutiMetricsCall call;

public int dataSize() {
return cacheData.size();
}

@PostConstruct
public void init() {
Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
try {
if (cacheData.size() > 4) {
log.info("clear cache data:{}", cacheData.size());
cacheData.clear();
}
} catch (Throwable ex) {
log.error(ex.getMessage());
}
}, 0, 60, TimeUnit.SECONDS);

Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
try {
enterManager.getMonitor().enter();
cacheData();
} catch (Throwable ex) {
log.error(ex.getMessage(), ex);
} finally {
enterManager.getMonitor().leave();
}
}, 0, 15, TimeUnit.SECONDS);
}

@Resource
private EnterManager enterManager;




public byte[] getData() {
log.info("get data");
Stopwatch sw = Stopwatch.createStarted();
try {
if (cacheData.size() >= 1) {
return cacheData.remove(0);
}
} finally {
log.info("get data use time:{}ms", sw.elapsed(TimeUnit.MILLISECONDS));
}
return new byte[]{};
}


public void cacheData() {
call.change();
Executors.newSingleThreadExecutor().submit(() -> {
log.info("cache data");
Stopwatch sw = Stopwatch.createStarted();
List<String> list = new ArrayList<>();
MutiMetrics old = call.old();
CollectorRegistry registry = old.getRegistry();
try {
Field field = registry.getClass().getDeclaredField("namesToCollectors");
field.setAccessible(true);
Map<String, Collector> namesToCollectors = (Map<String, Collector>) field.get(registry);
list = namesToCollectors.keySet().stream()
.filter(it -> !it.endsWith("created"))
.collect(Collectors.toList());
} catch (Exception e) {
log.info("export metrics error : ", e);
}
try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); OutputStreamWriter writer = new OutputStreamWriter(baos)) {
TextFormat.writeFormat(TextFormat.CONTENT_TYPE_004, writer, registry.filteredMetricFamilySamples(Sets.newHashSet(list)));
writer.flush();
byte[] bytes = baos.toByteArray();
this.cacheData.add(bytes);
} catch (Throwable ex) {
log.error(ex.getMessage());
} finally {
clearMetrics(old);
}
log.info("cache data use time:{} ms", sw.elapsed(TimeUnit.MILLISECONDS));
});
}


private void clearMetrics(MutiMetrics old) {
try {
MetricsManager gMetricsMgr = old.gMetricsMgr;
if (gMetricsMgr instanceof Prometheus) {
Prometheus prometheus = (Prometheus) gMetricsMgr;
Map<String, Object> prometheusMetrics = prometheus.prometheusMetrics;
clearTypeMetrics(prometheusMetrics,old.getRegistry());
prometheus.prometheusMetrics.clear();
prometheus.prometheusTypeMetrics.clear();
}
} catch (Exception e) {
log.error("clear metrics error", e);
}
}

private void clearTypeMetrics(Map<String, Object> prometheusMetrics, CollectorRegistry registry) {
for (String key : prometheusMetrics.keySet()) {
Object o = prometheusMetrics.get(key);
if (o instanceof Counter) {
Counter counter = (Counter) o;
registry.unregister(counter);
} else if (o instanceof Gauge) {
Gauge gauge = (Gauge) o;
gauge.clear();
registry.unregister(gauge);
} else if (o instanceof Histogram) {
Histogram histogram = (Histogram) o;
histogram.clear();
registry.unregister(histogram);
} else {
log.error("metrics : " + key + " Type conversion failed, original type : " + o.getClass().getName());
}
}
}


}
42 changes: 42 additions & 0 deletions ...e-etl/trace-etl-server/src/main/java/com/xiaomi/hera/trace/etl/consumer/EnterManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
package com.xiaomi.hera.trace.etl.consumer;

import com.google.common.util.concurrent.Monitor;
import lombok.Getter;
import org.springframework.stereotype.Service;

import java.util.function.BooleanSupplier;

/**
* @author goodjava@qq.com
* @date 2023/8/31 13:49
*/
@Service
public class EnterManager {

@Getter
private Monitor monitor = new Monitor();

@Getter
private Monitor processMonitor = new Monitor();

private Monitor.Guard guard = monitor.newGuard(new BooleanSupplier() {
@Override
public boolean getAsBoolean() {
return false;
}
});


public void enter() {
monitor.enter();
monitor.leave();
}

public void processEnter() {
processMonitor.enter();
processMonitor.leave();
}



}
Loading
Loading