-
Notifications
You must be signed in to change notification settings - Fork 999
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[ospp] add push style collector (#1222)
- Loading branch information
Showing
44 changed files
with
1,049 additions
and
84 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
166 changes: 166 additions & 0 deletions
166
collector/src/main/java/org/dromara/hertzbeat/collector/collect/push/PushCollectImpl.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,166 @@ | ||
| package org.dromara.hertzbeat.collector.collect.push; | ||
|
|
||
| import com.fasterxml.jackson.core.type.TypeReference; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.http.HttpHeaders; | ||
| import org.apache.http.HttpHost; | ||
| import org.apache.http.client.config.RequestConfig; | ||
| import org.apache.http.client.methods.CloseableHttpResponse; | ||
| import org.apache.http.client.methods.HttpUriRequest; | ||
| import org.apache.http.client.methods.RequestBuilder; | ||
| import org.apache.http.client.protocol.HttpClientContext; | ||
| import org.apache.http.protocol.HttpContext; | ||
| import org.apache.http.util.EntityUtils; | ||
| import org.dromara.hertzbeat.collector.collect.AbstractCollect; | ||
| import org.dromara.hertzbeat.collector.collect.common.http.CommonHttpClient; | ||
| import org.dromara.hertzbeat.collector.dispatch.DispatchConstants; | ||
| import org.dromara.hertzbeat.collector.util.CollectUtil; | ||
| import org.dromara.hertzbeat.common.constants.CollectorConstants; | ||
| import org.dromara.hertzbeat.common.entity.dto.Message; | ||
| import org.dromara.hertzbeat.common.entity.job.Metrics; | ||
| import org.dromara.hertzbeat.common.entity.job.protocol.PushProtocol; | ||
| import org.dromara.hertzbeat.common.entity.message.CollectRep; | ||
| import org.dromara.hertzbeat.common.entity.push.PushMetricsDto; | ||
| import org.dromara.hertzbeat.common.util.CommonUtil; | ||
| import org.dromara.hertzbeat.common.util.IpDomainUtil; | ||
| import org.dromara.hertzbeat.common.util.JsonUtil; | ||
|
|
||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| /** | ||
| * push style collect | ||
| * | ||
| * @author vinci | ||
| */ | ||
| @Slf4j | ||
| public class PushCollectImpl extends AbstractCollect { | ||
|
|
||
| private static Map<Long, Long> timeMap = new ConcurrentHashMap<>(); | ||
|
|
||
| // ms | ||
| private static final Integer timeout = 3000; | ||
|
|
||
| private static final Integer SUCCESS_CODE = 200; | ||
|
|
||
| // 第一次采集多久之前的数据，其实没有办法确定，因为无法确定上次何时采集，难以避免重启后重复采集的现象，默认30s | ||
| private static final Integer firstCollectInterval = 30000; | ||
|
|
||
| public PushCollectImpl() { | ||
| } | ||
|
|
||
| @Override | ||
| public void collect(CollectRep.MetricsData.Builder builder, | ||
| long appId, String app, Metrics metrics) { | ||
| long curTime = System.currentTimeMillis(); | ||
|
|
||
| PushProtocol pushProtocol = metrics.getPush(); | ||
|
|
||
| Long time = timeMap.getOrDefault(appId, curTime - firstCollectInterval); | ||
| timeMap.put(appId, curTime); | ||
|
|
||
| HttpContext httpContext = createHttpContext(pushProtocol); | ||
| HttpUriRequest request = createHttpRequest(pushProtocol, appId, time); | ||
|
|
||
| try { | ||
| CloseableHttpResponse response = CommonHttpClient.getHttpClient().execute(request, httpContext); | ||
| int statusCode = response.getStatusLine().getStatusCode(); | ||
| if (statusCode != SUCCESS_CODE) { | ||
| builder.setCode(CollectRep.Code.FAIL); | ||
| builder.setMsg("StatusCode " + statusCode); | ||
| return; | ||
| } | ||
| String resp = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8); | ||
|
|
||
| parseResponse(builder, resp, metrics); | ||
|
|
||
| } catch (Exception e) { | ||
| String errorMsg = CommonUtil.getMessageFromThrowable(e); | ||
| log.error(errorMsg, e); | ||
| builder.setCode(CollectRep.Code.FAIL); | ||
| builder.setMsg(errorMsg); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public String supportProtocol() { | ||
| return DispatchConstants.PROTOCOL_PUSH; | ||
| } | ||
|
|
||
| private HttpContext createHttpContext(PushProtocol pushProtocol) { | ||
| HttpHost host = new HttpHost(pushProtocol.getHost(), Integer.parseInt(pushProtocol.getPort())); | ||
| HttpClientContext httpClientContext = new HttpClientContext(); | ||
| httpClientContext.setTargetHost(host); | ||
| return httpClientContext; | ||
| } | ||
|
|
||
| private HttpUriRequest createHttpRequest(PushProtocol pushProtocol, Long monitorId, Long startTime) { | ||
| RequestBuilder requestBuilder = RequestBuilder.get(); | ||
|
|
||
|
|
||
| // uri | ||
| String uri = CollectUtil.replaceUriSpecialChar(pushProtocol.getUri()); | ||
| if (IpDomainUtil.isHasSchema(pushProtocol.getHost())) { | ||
| requestBuilder.setUri(pushProtocol.getHost() + ":" + pushProtocol.getPort() + uri); | ||
| } else { | ||
| String ipAddressType = IpDomainUtil.checkIpAddressType(pushProtocol.getHost()); | ||
| String baseUri = CollectorConstants.IPV6.equals(ipAddressType) | ||
| ? String.format("[%s]:%s", pushProtocol.getHost(), pushProtocol.getPort() + uri) | ||
| : String.format("%s:%s", pushProtocol.getHost(), pushProtocol.getPort() + uri); | ||
|
|
||
| requestBuilder.setUri(CollectorConstants.HTTP_HEADER + baseUri); | ||
| } | ||
|
|
||
| requestBuilder.addHeader(HttpHeaders.CONNECTION, "keep-alive"); | ||
| requestBuilder.addHeader(HttpHeaders.USER_AGENT, "Mozilla/5.0 (Windows NT 6.1; WOW64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2272.76 Safari/537.36"); | ||
|
|
||
| requestBuilder.addParameter("id", String.valueOf(monitorId)); | ||
| requestBuilder.addParameter("time", String.valueOf(startTime)); | ||
| requestBuilder.addHeader(HttpHeaders.ACCEPT, "application/json"); | ||
|
|
||
|
|
||
| //requestBuilder.setUri(pushProtocol.getUri()); | ||
|
|
||
| if (timeout > 0) { | ||
| RequestConfig requestConfig = RequestConfig.custom() | ||
| .setConnectTimeout(timeout) | ||
| .setSocketTimeout(timeout) | ||
| .setRedirectsEnabled(true) | ||
| .build(); | ||
| requestBuilder.setConfig(requestConfig); | ||
| } | ||
|
|
||
| return requestBuilder.build(); | ||
| } | ||
|
|
||
| private void parseResponse(CollectRep.MetricsData.Builder builder, String resp, Metrics metric) { | ||
| // Map<String, Object> jsonMap = JsonUtil.fromJson(resp, new TypeReference<Map<String, Object>>() { | ||
| // }); | ||
| // if (jsonMap == null) { | ||
| // throw new NullPointerException("parse result is null"); | ||
| // } | ||
| Message<PushMetricsDto> msg = JsonUtil.fromJson(resp, new TypeReference<Message<PushMetricsDto>>() { | ||
| }); | ||
| if (msg == null) { | ||
| throw new NullPointerException("parse result is null"); | ||
| } | ||
| PushMetricsDto pushMetricsDto = msg.getData(); | ||
| if (pushMetricsDto == null || pushMetricsDto.getMetricsList() == null) { | ||
| throw new NullPointerException("parse result is null"); | ||
| } | ||
| for (PushMetricsDto.Metrics pushMetrics : pushMetricsDto.getMetricsList()) { | ||
| List<String> metricColumn = new ArrayList<>(); | ||
| for (Metrics.Field field : metric.getFields()) { | ||
| metricColumn.add(pushMetrics.getMetrics().get(field.getField())); | ||
| } | ||
| CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder() | ||
| .addAllColumns(metricColumn); | ||
| builder.addValues(valueRowBuilder.build()); | ||
| } | ||
| builder.setTime(System.currentTimeMillis()); | ||
| } | ||
| } |
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
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
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.