-
Notifications
You must be signed in to change notification settings - Fork 154
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #597 from goodjava/master
etl_update
- Loading branch information
Showing
12 changed files
with
443 additions
and
110 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
117 changes: 117 additions & 0 deletions
117
...l/trace-etl-server/src/main/java/com/xiaomi/hera/trace/etl/consumer/DataCacheService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| package com.xiaomi.hera.trace.etl.consumer; | ||
|
|
||
| import com.google.common.base.Stopwatch; | ||
| import com.google.common.collect.Sets; | ||
| import com.xiaomi.hera.trace.etl.constant.LockUtil; | ||
| import com.xiaomi.youpin.prometheus.client.Metrics; | ||
| import com.xiaomi.youpin.prometheus.client.MetricsManager; | ||
| import com.xiaomi.youpin.prometheus.client.Prometheus; | ||
| import io.prometheus.client.*; | ||
| import io.prometheus.client.exporter.common.TextFormat; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import okio.Buffer; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| 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.TimeUnit; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * @author [email protected] | ||
| * @date 2023/8/29 10:02 | ||
| */ | ||
| @Service | ||
| @Slf4j | ||
| public class DataCacheService { | ||
|
|
||
| private CopyOnWriteArrayList<byte[]> data = new CopyOnWriteArrayList<>(); | ||
|
|
||
| public byte[] getData() { | ||
| log.info("get data"); | ||
| Stopwatch sw = Stopwatch.createStarted(); | ||
| Buffer buffer = new Buffer(); | ||
| try { | ||
| data.forEach(it -> buffer.write(it)); | ||
| data.clear(); | ||
| return buffer.readByteArray(); | ||
| } finally { | ||
| log.info("get data use time:{}ms", sw.elapsed(TimeUnit.MILLISECONDS)); | ||
| buffer.clear(); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| public void cacheData() { | ||
| log.info("cache data"); | ||
| Stopwatch sw = Stopwatch.createStarted(); | ||
| synchronized (LockUtil.lock) { | ||
| List<String> list = new ArrayList<>(); | ||
| CollectorRegistry registry = CollectorRegistry.defaultRegistry; | ||
| 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.data.add(bytes); | ||
| } catch (Throwable ex) { | ||
| log.error(ex.getMessage()); | ||
| } finally { | ||
| clearMetrics(); | ||
| } | ||
| } | ||
| log.info("cache data use time:{}ms", sw.elapsed(TimeUnit.MILLISECONDS)); | ||
| } | ||
|
|
||
|
|
||
| private void clearMetrics() { | ||
| try { | ||
| MetricsManager gMetricsMgr = Metrics.getInstance().gMetricsMgr; | ||
| if (gMetricsMgr instanceof Prometheus) { | ||
| Prometheus prometheus = (Prometheus) gMetricsMgr; | ||
| Map<String, Object> prometheusMetrics = prometheus.prometheusMetrics; | ||
| clearTypeMetrics(prometheusMetrics); | ||
| prometheus.prometheusMetrics.clear(); | ||
| prometheus.prometheusTypeMetrics.clear(); | ||
| } | ||
| } catch (Exception e) { | ||
| log.error("clear metrics error", e); | ||
| } | ||
| } | ||
|
|
||
| private void clearTypeMetrics(Map<String, Object> prometheusMetrics) { | ||
| for (String key : prometheusMetrics.keySet()) { | ||
| Object o = prometheusMetrics.get(key); | ||
| if (o instanceof Counter) { | ||
| Counter counter = (Counter) o; | ||
| CollectorRegistry.defaultRegistry.unregister(counter); | ||
| } else if (o instanceof Gauge) { | ||
| Gauge gauge = (Gauge) o; | ||
| gauge.clear(); | ||
| CollectorRegistry.defaultRegistry.unregister(gauge); | ||
| } else if (o instanceof Histogram) { | ||
| Histogram histogram = (Histogram) o; | ||
| histogram.clear(); | ||
| CollectorRegistry.defaultRegistry.unregister(histogram); | ||
| } else { | ||
| log.error("metrics : " + key + " Type conversion failed, original type : " + o.getClass().getName()); | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.