mirror of
https://github.com/dromara/hertzbeat.git
synced 2026-09-17 09:40:58 +00:00
[bugfix] Remove broken Push Style Monitor (app-push) support (#4339)
Co-authored-by: aias00 <liuhongyu@apache.org>
This commit is contained in:
Prabal Pratap Singh
co-authored by
aias00
aias00
parent
6671f4603b
commit
dffb38b4c5
-183
@@ -1,183 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.collector.collect.push;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.hertzbeat.collector.collect.AbstractCollect;
|
||||
import org.apache.hertzbeat.collector.collect.common.http.CommonHttpClient;
|
||||
import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
|
||||
import org.apache.hertzbeat.collector.util.CollectUtil;
|
||||
import org.apache.hertzbeat.common.constants.NetworkConstants;
|
||||
import org.apache.hertzbeat.common.constants.SignConstants;
|
||||
import org.apache.hertzbeat.common.entity.dto.Message;
|
||||
import org.apache.hertzbeat.common.entity.job.Metrics;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.PushProtocol;
|
||||
import org.apache.hertzbeat.common.entity.message.CollectRep;
|
||||
import org.apache.hertzbeat.common.entity.push.PushMetricsDto;
|
||||
import org.apache.hertzbeat.common.util.CommonUtil;
|
||||
import org.apache.hertzbeat.common.util.IpDomainUtil;
|
||||
import org.apache.hertzbeat.common.util.JsonUtil;
|
||||
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.springframework.http.MediaType;
|
||||
import tools.jackson.core.type.TypeReference;
|
||||
|
||||
/**
|
||||
* push style collect
|
||||
*/
|
||||
@Slf4j
|
||||
public class PushCollectImpl extends AbstractCollect {
|
||||
|
||||
private static final Map<Long, Long> timeMap = new ConcurrentHashMap<>();
|
||||
|
||||
// ms
|
||||
private static final Integer DEFAULT_TIMEOUT = 3000;
|
||||
|
||||
private static final Integer SUCCESS_CODE = 200;
|
||||
|
||||
// It's hard to determine how long ago the first data collection was, because there's no way to know when the last collection occurred.
|
||||
// This makes it difficult to avoid re-collecting data after a restart. The default is 30 seconds
|
||||
private static final Integer FIRST_COLLECT_INTERVAL = 30000;
|
||||
|
||||
@Override
|
||||
public void preCheck(Metrics metrics) throws IllegalArgumentException {
|
||||
if (metrics == null || metrics.getPush() == null) {
|
||||
throw new IllegalArgumentException("Push collect must has Push params");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void collect(CollectRep.MetricsData.Builder builder,
|
||||
Metrics metrics) {
|
||||
long curTime = System.currentTimeMillis();
|
||||
long monitorId = builder.getId();
|
||||
PushProtocol pushProtocol = metrics.getPush();
|
||||
|
||||
Long time = timeMap.getOrDefault(monitorId, curTime - FIRST_COLLECT_INTERVAL);
|
||||
timeMap.put(monitorId, curTime);
|
||||
|
||||
HttpContext httpContext = createHttpContext(pushProtocol);
|
||||
HttpUriRequest request = createHttpRequest(pushProtocol, monitorId, 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(NetworkConstants.STATUS_CODE + SignConstants.BLANK + 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 = NetworkConstants.IPV6.equals(ipAddressType)
|
||||
? String.format("[%s]:%s", pushProtocol.getHost(), pushProtocol.getPort() + uri)
|
||||
: String.format("%s:%s", pushProtocol.getHost(), pushProtocol.getPort() + uri);
|
||||
|
||||
requestBuilder.setUri(NetworkConstants.HTTP_HEADER + baseUri);
|
||||
}
|
||||
|
||||
requestBuilder.addHeader(HttpHeaders.CONNECTION, NetworkConstants.KEEP_ALIVE);
|
||||
requestBuilder.addHeader(HttpHeaders.USER_AGENT, NetworkConstants.USER_AGENT);
|
||||
|
||||
requestBuilder.addParameter("id", String.valueOf(monitorId));
|
||||
requestBuilder.addParameter("time", String.valueOf(startTime));
|
||||
requestBuilder.addHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE);
|
||||
|
||||
|
||||
//requestBuilder.setUri(pushProtocol.getUri());
|
||||
|
||||
if (DEFAULT_TIMEOUT > 0) {
|
||||
RequestConfig requestConfig = RequestConfig.custom()
|
||||
.setConnectTimeout(DEFAULT_TIMEOUT)
|
||||
.setSocketTimeout(DEFAULT_TIMEOUT)
|
||||
.setRedirectsEnabled(true)
|
||||
.build();
|
||||
requestBuilder.setConfig(requestConfig);
|
||||
}
|
||||
|
||||
return requestBuilder.build();
|
||||
}
|
||||
|
||||
private void parseResponse(CollectRep.MetricsData.Builder builder, String resp, Metrics metric) {
|
||||
Message<PushMetricsDto> msg = JsonUtil.fromJson(resp, new TypeReference<>() {
|
||||
});
|
||||
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()) {
|
||||
for (Map<String, String> metrics : pushMetrics.getMetrics()) {
|
||||
List<String> metricColumn = new ArrayList<>();
|
||||
for (Metrics.Field field : metric.getFields()) {
|
||||
metricColumn.add(metrics.get(field.getField()));
|
||||
}
|
||||
CollectRep.ValueRow valueRow = CollectRep.ValueRow.newBuilder()
|
||||
.addAllColumns(metricColumn).build();
|
||||
builder.addValueRow(valueRow);
|
||||
}
|
||||
}
|
||||
builder.setTime(System.currentTimeMillis());
|
||||
}
|
||||
}
|
||||
-72
@@ -1,72 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.collector.collect.push;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
|
||||
import org.apache.hertzbeat.common.entity.job.Metrics;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.PushProtocol;
|
||||
import org.apache.hertzbeat.common.entity.message.CollectRep;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* Test case for {@link PushCollectImpl}
|
||||
*/
|
||||
public class PushCollectImplTest {
|
||||
private PushCollectImpl pushCollect;
|
||||
private PushProtocol push;
|
||||
private CollectRep.MetricsData.Builder builder;
|
||||
|
||||
@BeforeEach
|
||||
public void setup() {
|
||||
pushCollect = new PushCollectImpl();
|
||||
push = PushProtocol.builder().uri("/metrics").host("example.com").port("60").build();
|
||||
builder = CollectRep.MetricsData.newBuilder();
|
||||
}
|
||||
|
||||
@Test
|
||||
void preCheck() throws Exception {
|
||||
// metrics is null
|
||||
assertThrows(IllegalArgumentException.class, () -> pushCollect.preCheck(null));
|
||||
|
||||
// protocol is null
|
||||
assertThrows(IllegalArgumentException.class, () -> pushCollect.preCheck(new Metrics()));
|
||||
|
||||
// everyting is ok
|
||||
assertDoesNotThrow(() -> {
|
||||
pushCollect.preCheck(Metrics.builder().push(push).build());
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void collect() throws Exception {
|
||||
assertDoesNotThrow(() -> {
|
||||
pushCollect.collect(builder, Metrics.builder().push(push).build());
|
||||
assertEquals(CollectRep.Code.FAIL, builder.getCode());
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void supportProtocol() {
|
||||
assertEquals(DispatchConstants.PROTOCOL_PUSH, pushCollect.supportProtocol());
|
||||
}
|
||||
}
|
||||
-1
@@ -14,7 +14,6 @@ org.apache.hertzbeat.collector.collect.ntp.NtpCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.websocket.WebsocketCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.ftp.FtpCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.udp.UdpCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.push.PushCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.dns.DnsCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.nginx.NginxCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.memcached.MemcachedCollectImpl
|
||||
|
||||
-1
@@ -14,7 +14,6 @@ org.apache.hertzbeat.collector.collect.ntp.NtpCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.websocket.WebsocketCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.ftp.FtpCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.udp.UdpCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.push.PushCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.dns.DnsCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.nginx.NginxCollectImpl
|
||||
org.apache.hertzbeat.collector.collect.memcached.MemcachedCollectImpl
|
||||
|
||||
-4
@@ -103,10 +103,6 @@ public interface DispatchConstants {
|
||||
* protocol rocketmq
|
||||
*/
|
||||
String PROTOCOL_ROCKETMQ = "rocketmq";
|
||||
/**
|
||||
* protocol push
|
||||
*/
|
||||
String PROTOCOL_PUSH = "push";
|
||||
/**
|
||||
* protocol prometheus
|
||||
*/
|
||||
|
||||
-4
@@ -20,7 +20,6 @@ package org.apache.hertzbeat.collector.timer;
|
||||
import com.google.gson.Gson;
|
||||
import com.google.gson.JsonElement;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
|
||||
import org.apache.hertzbeat.collector.dispatch.MetricsTaskDispatch;
|
||||
import org.apache.hertzbeat.collector.util.CollectUtil;
|
||||
import org.apache.hertzbeat.common.constants.CommonConstants;
|
||||
@@ -84,9 +83,6 @@ public class WheelTimerTask implements TimerTask {
|
||||
JsonElement jsonElement = GSON.toJsonTree(metric);
|
||||
CollectUtil.replaceSmilingPlaceholder(jsonElement, configmap);
|
||||
metric = GSON.fromJson(jsonElement, Metrics.class);
|
||||
if (job.getApp().equals(DispatchConstants.PROTOCOL_PUSH)) {
|
||||
CollectUtil.replaceFieldsForPushStyleMonitor(metric, configmap);
|
||||
}
|
||||
metricsTmp.add(metric);
|
||||
}
|
||||
job.setMetrics(metricsTmp);
|
||||
|
||||
-7
@@ -437,13 +437,6 @@ public final class CollectUtil {
|
||||
return mapList;
|
||||
}
|
||||
|
||||
public static void replaceFieldsForPushStyleMonitor(Metrics metrics, Map<String, Configmap> configmap) {
|
||||
|
||||
List<Metrics.Field> pushFieldList = JsonUtil.fromJson((String) configmap.get("fields").getValue(), new TypeReference<>() {
|
||||
});
|
||||
metrics.setFields(pushFieldList);
|
||||
}
|
||||
|
||||
/**
|
||||
* convert 16 hexString to byte[]
|
||||
* eg: 302c0201010409636f6d6d756e697479a11c020419e502e7020100020100300e300c06082b060102010102000500
|
||||
|
||||
-5
@@ -54,7 +54,6 @@ import org.apache.hertzbeat.common.entity.job.protocol.NgqlProtocol;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.NtpProtocol;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.Pop3Protocol;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.PrometheusProtocol;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.PushProtocol;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.RedfishProtocol;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.RedisProtocol;
|
||||
import org.apache.hertzbeat.common.entity.job.protocol.RocketmqProtocol;
|
||||
@@ -212,10 +211,6 @@ public class Metrics {
|
||||
* Monitoring configuration information using the public rocketmq protocol
|
||||
*/
|
||||
private RocketmqProtocol rocketmq;
|
||||
/**
|
||||
* Monitoring configuration information using push style
|
||||
*/
|
||||
private PushProtocol push;
|
||||
/**
|
||||
* Monitoring configuration information using the public prometheus protocol
|
||||
*/
|
||||
|
||||
-69
@@ -1,69 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.common.entity.job.protocol;
|
||||
|
||||
import static org.apache.hertzbeat.common.util.IpDomainUtil.isHasSchema;
|
||||
import static org.apache.hertzbeat.common.util.IpDomainUtil.validPort;
|
||||
import static org.apache.hertzbeat.common.util.IpDomainUtil.validateIpDomain;
|
||||
|
||||
import java.util.List;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.apache.hertzbeat.common.entity.dto.Field;
|
||||
|
||||
/**
|
||||
* push protocol definition
|
||||
*/
|
||||
@Data
|
||||
@Builder
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public class PushProtocol implements CommonRequestProtocol, Protocol {
|
||||
private String host;
|
||||
private String port;
|
||||
private String uri = "/api/push";
|
||||
private List<Field> fields;
|
||||
|
||||
@Override
|
||||
public boolean isInvalid() {
|
||||
if ((!validateIpDomain(host) && !isHasSchema(host)) || !validPort(port)) {
|
||||
return true;
|
||||
}
|
||||
if (Integer.parseInt(port) <= 0) {
|
||||
return true;
|
||||
}
|
||||
if (StringUtils.isBlank(uri) || !uri.startsWith("/") || StringUtils.containsWhitespace(uri)) {
|
||||
return true;
|
||||
}
|
||||
if (fields == null || fields.isEmpty()) {
|
||||
return true;
|
||||
}
|
||||
for (Field field : fields) {
|
||||
if (field == null
|
||||
|| StringUtils.isBlank(field.getName())
|
||||
|| field.getType() == null
|
||||
|| (field.getType() != 0 && field.getType() != 1)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
-54
@@ -1,54 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.common.entity.push;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/**
|
||||
* push metrics dto
|
||||
*/
|
||||
@Data
|
||||
@Builder
|
||||
@AllArgsConstructor
|
||||
public class PushMetricsDto {
|
||||
|
||||
List<Metrics> metricsList;
|
||||
|
||||
public PushMetricsDto() {
|
||||
metricsList = new ArrayList<>();
|
||||
}
|
||||
|
||||
/**
|
||||
* metrics
|
||||
*/
|
||||
@Data
|
||||
@Builder
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public static class Metrics {
|
||||
private long monitorId;
|
||||
private Long time;
|
||||
private List<Map<String, String>> metrics;
|
||||
}
|
||||
}
|
||||
-138
@@ -1,138 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hertzbeat.common.entity.job.protocol;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import java.util.List;
|
||||
import org.apache.hertzbeat.common.entity.dto.Field;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class PushProtocolTest {
|
||||
|
||||
@Test
|
||||
void isInvalidValidProtocol() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("/api/push")
|
||||
.fields(List.of(Field.builder().name("cpuUsage").type((byte) 0).build()))
|
||||
.build();
|
||||
assertFalse(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidValidProtocolWithSchemaHost() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("http://127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("/api/push")
|
||||
.fields(List.of(Field.builder().name("status").type((byte) 1).build()))
|
||||
.build();
|
||||
assertFalse(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidInvalidHost() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("")
|
||||
.port("1157")
|
||||
.uri("/api/push")
|
||||
.fields(List.of(Field.builder().name("status").type((byte) 1).build()))
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidZeroPort() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("0")
|
||||
.uri("/api/push")
|
||||
.fields(List.of(Field.builder().name("status").type((byte) 1).build()))
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidBlankUri() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("")
|
||||
.fields(List.of(Field.builder().name("status").type((byte) 1).build()))
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidUriWithoutLeadingSlash() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("api/push")
|
||||
.fields(List.of(Field.builder().name("status").type((byte) 1).build()))
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidBlankFields() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("/api/push")
|
||||
.fields(List.of())
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidFieldWithoutName() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("/api/push")
|
||||
.fields(List.of(Field.builder().name("").type((byte) 1).build()))
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidFieldWithoutType() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("/api/push")
|
||||
.fields(List.of(Field.builder().name("status").type(null).build()))
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInvalidFieldWithUnsupportedType() {
|
||||
PushProtocol protocol = PushProtocol.builder()
|
||||
.host("127.0.0.1")
|
||||
.port("1157")
|
||||
.uri("/api/push")
|
||||
.fields(List.of(Field.builder().name("status").type((byte) 2).build()))
|
||||
.build();
|
||||
assertTrue(protocol.isInvalid());
|
||||
}
|
||||
}
|
||||
-8
@@ -69,14 +69,6 @@ public class AppController {
|
||||
return ResponseUtil.handle(() -> appService.getAppParamDefines(app.toLowerCase()));
|
||||
}
|
||||
|
||||
@GetMapping(path = "/{monitorId}/pushdefine")
|
||||
@Operation(summary = "The definition structure of the specified monitoring type according to the push query",
|
||||
description = "The definition structure of the specified monitoring type according to the push query")
|
||||
public ResponseEntity<Message<Job>> queryPushDefine(
|
||||
@Parameter(description = "en: Monitoring type name", example = "api") @PathVariable("monitorId") final Long monitorId) {
|
||||
return ResponseUtil.handle(() -> appService.getPushDefine(monitorId));
|
||||
}
|
||||
|
||||
@GetMapping(path = "/{monitorId}/define/dynamic")
|
||||
@Operation(summary = "The definition structure of the specified monitoring type according to the push query",
|
||||
description = "The definition structure of the specified monitoring type according to the push query")
|
||||
|
||||
@@ -37,8 +37,6 @@ public interface AppService {
|
||||
*/
|
||||
List<ParamDefineInfo> getAppParamDefines(String app);
|
||||
|
||||
Job getPushDefine(Long monitorId);
|
||||
|
||||
/**
|
||||
* get auto generate dynamic template define
|
||||
* for prometheus and more type
|
||||
|
||||
-31
@@ -22,14 +22,11 @@ import org.apache.commons.io.FileUtils;
|
||||
import org.apache.commons.io.IOUtils;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
|
||||
import org.apache.hertzbeat.collector.util.CollectUtil;
|
||||
import org.apache.hertzbeat.common.entity.job.Configmap;
|
||||
import org.apache.hertzbeat.common.entity.job.Job;
|
||||
import org.apache.hertzbeat.common.entity.job.Metrics;
|
||||
import org.apache.hertzbeat.common.entity.job.RuntimeParamDefine;
|
||||
import org.apache.hertzbeat.common.entity.manager.Define;
|
||||
import org.apache.hertzbeat.common.entity.manager.Monitor;
|
||||
import org.apache.hertzbeat.common.entity.manager.Param;
|
||||
import org.apache.hertzbeat.common.entity.message.CollectRep;
|
||||
import org.apache.hertzbeat.common.util.CommonUtil;
|
||||
import org.apache.hertzbeat.common.util.HertzBeatKeywordsUtil;
|
||||
@@ -87,8 +84,6 @@ import static java.util.Objects.isNull;
|
||||
@Slf4j
|
||||
public class AppServiceImpl implements AppService, InitializingBean {
|
||||
|
||||
private static final String PUSH_PROTOCOL_METRICS_NAME = "metrics";
|
||||
|
||||
private final MonitorDao monitorDao;
|
||||
private final ObjectStoreConfigServiceImpl objectStoreConfigService;
|
||||
private final ParamDao paramDao;
|
||||
@@ -131,29 +126,6 @@ public class AppServiceImpl implements AppService, InitializingBean {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Job getPushDefine(Long monitorId) throws IllegalArgumentException {
|
||||
Job appDefine = appDefines.get(DispatchConstants.PROTOCOL_PUSH);
|
||||
if (appDefine == null) {
|
||||
throw new IllegalArgumentException("The push collector not support.");
|
||||
}
|
||||
List<Metrics> metrics = appDefine.getMetrics();
|
||||
List<Metrics> metricsTmp = new ArrayList<>();
|
||||
for (Metrics metric : metrics) {
|
||||
if (PUSH_PROTOCOL_METRICS_NAME.equals(metric.getName())) {
|
||||
List<Param> params = paramDao.findParamsByMonitorId(monitorId);
|
||||
List<Configmap> configmaps = params.stream()
|
||||
.map(param -> new Configmap(param.getField(), param.getParamValue(),
|
||||
param.getType())).toList();
|
||||
Map<String, Configmap> configmap = configmaps.stream().collect(Collectors.toMap(Configmap::getKey, item -> item, (key1, key2) -> key1));
|
||||
CollectUtil.replaceFieldsForPushStyleMonitor(metric, configmap);
|
||||
metricsTmp.add(metric);
|
||||
}
|
||||
}
|
||||
appDefine.setMetrics(metricsTmp);
|
||||
return appDefine;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Job getAutoGenerateDynamicDefine(Long monitorId) {
|
||||
Job job = getAppDefine(DispatchConstants.PROTOCOL_PROMETHEUS);
|
||||
@@ -293,9 +265,6 @@ public class AppServiceImpl implements AppService, InitializingBean {
|
||||
public List<Hierarchy> getAppHierarchy(String app, String lang) {
|
||||
LinkedList<Hierarchy> hierarchies = new LinkedList<>();
|
||||
Job job = appDefines.get(app.toLowerCase());
|
||||
if (DispatchConstants.PROTOCOL_PUSH.equalsIgnoreCase(job.getApp())) {
|
||||
return hierarchies;
|
||||
}
|
||||
queryAppHierarchy(lang, hierarchies, job);
|
||||
return hierarchies;
|
||||
}
|
||||
|
||||
@@ -1,85 +0,0 @@
|
||||
# Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
# contributor license agreements. See the NOTICE file distributed with
|
||||
# this work for additional information regarding copyright ownership.
|
||||
# The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
# (the "License"); you may not use this file except in compliance with
|
||||
# the License. You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
# The monitoring type category：service-application service monitoring db-database monitoring custom-custom monitoring os-operating system monitoring
|
||||
category: __system__
|
||||
# Monitoring application type name (consistent with file name) eg: linux windows tomcat mysql aws...
|
||||
app: push
|
||||
# The app api i18n name
|
||||
name:
|
||||
zh-CN: 推送方式监控
|
||||
en-US: Push Style Monitor
|
||||
ja-JP: プッシュ方法のモニター
|
||||
# Input params define for app api(render web ui by the definition)
|
||||
params:
|
||||
# field-param field key
|
||||
- field: host
|
||||
# name-param field display i18n name
|
||||
name:
|
||||
zh-CN: 推送模块Host
|
||||
en-US: Push Module Host
|
||||
ja-JP: プッシュモジュールのホスト
|
||||
# type-param field type(most mapping the html input type)
|
||||
type: host
|
||||
# required-true or false
|
||||
required: true
|
||||
# field-param field key
|
||||
defaultValue: 127.0.0.1
|
||||
- field: port
|
||||
# name-param field display i18n name
|
||||
name:
|
||||
zh-CN: 端口
|
||||
en-US: Port
|
||||
ja-JP: ポート
|
||||
# type-param field type(most mapping the html input type)
|
||||
type: number
|
||||
# when type is number, range is required
|
||||
range: '[0,65535]'
|
||||
# required-true or false
|
||||
required: true
|
||||
# default value
|
||||
defaultValue: 1157
|
||||
- field: fields
|
||||
# name-param field display i18n name
|
||||
name:
|
||||
zh-CN: 监控数据字段
|
||||
en-US: Metrics fields
|
||||
ja-JP: メトリクスフィールド
|
||||
# type-param field type(most mapping the html input type)
|
||||
type: metrics-field
|
||||
# required-true or false
|
||||
required: true
|
||||
|
||||
# collect metrics config list
|
||||
metrics:
|
||||
# metrics - all
|
||||
- name: metrics
|
||||
i18n:
|
||||
zh-CN: 指标
|
||||
en-US: Metrics
|
||||
ja-JP: メトリクス
|
||||
# metrics scheduling priority(0->127)->(high->low), metrics with the same priority will be scheduled in parallel
|
||||
# priority 0's metrics is availability metrics, it will be scheduled first, only availability metrics collect success will the scheduling continue
|
||||
priority: 0
|
||||
# the protocol used for monitoring, eg: sql, ssh, http, telnet, wmi, snmp, sdk
|
||||
protocol: push
|
||||
# the config content when protocol is http
|
||||
push:
|
||||
# http host: ipv4 ipv6 domain
|
||||
host: ^_^host^_^
|
||||
# http port
|
||||
port: ^_^port^_^
|
||||
# http uri
|
||||
uri: /api/push
|
||||
-24
@@ -87,30 +87,6 @@ class AppControllerTest {
|
||||
.andReturn();
|
||||
}
|
||||
|
||||
@Test
|
||||
void queryPushDefine() throws Exception {
|
||||
// Data to make
|
||||
Job mockJob = new Job();
|
||||
mockJob.setId(1L);
|
||||
mockJob.setMonitorId(1L);
|
||||
mockJob.setCategory("os");
|
||||
mockJob.setApp("mac");
|
||||
mockJob.setName(new HashMap<>());
|
||||
mockJob.setMetrics(new ArrayList<>());
|
||||
mockJob.setConfigmap(new ArrayList<>());
|
||||
|
||||
|
||||
// The interface is called to return manufactured data
|
||||
Mockito.when(appService.getPushDefine(1L)).thenReturn(mockJob);
|
||||
|
||||
// Request interface
|
||||
this.mockMvc.perform(MockMvcRequestBuilders.get("/api/apps/{monitorId}/pushdefine", 1L))
|
||||
.andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
|
||||
.andExpect(jsonPath("$.data.id").value(1))
|
||||
.andExpect(jsonPath("$.data.app").value("mac"))
|
||||
.andReturn();
|
||||
}
|
||||
|
||||
@Test
|
||||
void queryAutoGenerateDynamicAppDefine() throws Exception {
|
||||
// Data to make
|
||||
|
||||
@@ -1,26 +0,0 @@
|
||||
---
|
||||
id: extend-push
|
||||
title: Push Style Custom Monitoring
|
||||
sidebar_label: Push Style Custom Monitoring
|
||||
---
|
||||
|
||||
> Push style curstom monitor is a type of monitor which allow user to configure metrics format and push metrics to hertzbeat with their own service.
|
||||
> Here we will introduce how to use this feature.
|
||||
|
||||
## Push style custom monitor collection process
|
||||
|
||||
【Peer Server Start Pushing Metrics】 -> 【HertzBeat Push Module Stage Metrics】-> 【HertzBeat Collect Module collect Metrics Periodically】
|
||||
|
||||
### Data parsing method
|
||||
|
||||
HertzBeat will parsing metrics with the format configured by user while adding new monitor.
|
||||
|
||||
### Create Monitor Steps
|
||||
|
||||
HertzBeat DashBoard -> Service Monitor -> Push Style Monitor -> New Push Style Monitor -> set Push Module Host (hertzbeat server ip, usually 127.0.0.1/localhost) -> set Push Module Port (hertzbeat server port, usually 1157) -> configure metrics field (unit: string, type: 0 number / 1 string) -> end
|
||||
|
||||
---
|
||||
|
||||
### Monitor Configuration Example
|
||||
|
||||
|
||||
@@ -1,26 +0,0 @@
|
||||
---
|
||||
id: extend-push
|
||||
title: Push Style Custom Monitoring
|
||||
sidebar_label: Push Style Custom Monitoring
|
||||
---
|
||||
|
||||
> 推送方式监控是一种特殊的监控，允许用户配置数据格式并编写代码将指标推送到 HertzBeat。
|
||||
> 下面我们将介绍如何使用这一功能。
|
||||
|
||||
## 推送方式监控的采集流程
|
||||
|
||||
【用户开始推送数据】->【HertzBeat推送模块暂存数据】->【HertzBeat采集模块定期采集数据】
|
||||
|
||||
### 数据解析方式
|
||||
|
||||
HertzBeat会使用用户添加新监控时配置的格式来解析数据。
|
||||
|
||||
### 创建监控步骤
|
||||
|
||||
HertzBeat页面 -> 应用服务监控 -> 推送方式监控 -> 新建推送方式监视器 -> 设置推送模块主机（HertzBeat服务器ip，通常为127.0.0.1或localhost） -> 设置推送模块端口（hertzbeat服务器端口，通常为1157） -> 配置数据字段（单位：字符串表示，类型：0表示数字/1表示字符串）-> 结束
|
||||
|
||||
---
|
||||
|
||||
### 监控配置示例
|
||||
|
||||
|
||||
@@ -71,7 +71,7 @@ export class StartupService {
|
||||
i18n: `monitor.app.${app}`
|
||||
});
|
||||
} else {
|
||||
if (app != 'prometheus' && app != 'push') {
|
||||
if (app != 'prometheus') {
|
||||
this.menuService.getItem('monitoring')?.children?.push({
|
||||
text: app,
|
||||
link: `/monitors?app=${app}`,
|
||||
|
||||
@@ -116,9 +116,7 @@ export class MonitorDetailComponent implements OnInit, OnDestroy {
|
||||
switchMap((message: Message<any>) => {
|
||||
if (message.code == 0) {
|
||||
// Filter the numerical metrics that can be aggregated under this monitor
|
||||
if (this.app == 'push') {
|
||||
return this.appDefineSvc.getPushDefine(this.monitorId);
|
||||
} else if (this.app == 'prometheus') {
|
||||
if (this.app == 'prometheus') {
|
||||
return this.appDefineSvc.getAppDynamicDefine(this.monitorId);
|
||||
} else {
|
||||
return this.appDefineSvc.getAppDefine(this.app);
|
||||
@@ -517,9 +515,7 @@ export class MonitorDetailComponent implements OnInit, OnDestroy {
|
||||
.pipe(
|
||||
switchMap((message: Message<any>) => {
|
||||
if (message.code == 0) {
|
||||
if (this.app == 'push') {
|
||||
return this.appDefineSvc.getPushDefine(this.monitorId);
|
||||
} else if (this.app == 'prometheus') {
|
||||
if (this.app == 'prometheus') {
|
||||
return this.appDefineSvc.getAppDynamicDefine(this.monitorId);
|
||||
} else {
|
||||
return this.appDefineSvc.getAppDefine(this.app);
|
||||
|
||||
@@ -40,13 +40,6 @@ export class AppDefineService {
|
||||
return this.http.get<Message<ParamDefine[]>>(paramDefineUri);
|
||||
}
|
||||
|
||||
public getPushDefine(monitorId: number | undefined | null): Observable<Message<any>> {
|
||||
if (monitorId === null || monitorId === undefined) {
|
||||
console.log('getPushDefine monitorId can not null');
|
||||
}
|
||||
return this.http.get<Message<any>>(`/apps/${monitorId}/pushdefine`);
|
||||
}
|
||||
|
||||
public getAppDynamicDefine(monitorId: number | undefined | null): Observable<Message<any>> {
|
||||
if (monitorId === null || monitorId === undefined) {
|
||||
console.log('getAppDynamicDefine monitorId can not null');
|
||||
|
||||
Reference in New Issue
Block a user