[feature] Support DNS Service Discovery (#3328)

Signed-off-by: Sherlock Yin <sherlock.yin1994@gmail.com> Co-authored-by: yinyijun <yinyijun6@mgtv.com> Co-authored-by: yinyijun <yingey2011> Co-authored-by: tomsun28 <tomsun28@outlook.com>
2026-09-17 09:40:58 +00:00 · 2025-05-15 08:25:50 +08:00
parent ca3348adf2
commit 7779056983
16 changed files with 358 additions and 7 deletions
@@ -0,0 +1,181 @@
+/*
+ * 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.sd;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.collect.AbstractCollect;
+import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.util.CommonUtil;
+import org.xbill.DNS.AAAARecord;
+import org.xbill.DNS.ARecord;
+import org.xbill.DNS.Lookup;
+import org.xbill.DNS.MXRecord;
+import org.xbill.DNS.NSRecord;
+import org.xbill.DNS.Record;
+import org.xbill.DNS.SRVRecord;
+import org.xbill.DNS.SimpleResolver;
+import org.xbill.DNS.TextParseException;
+import org.xbill.DNS.Type;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+/**
+ * DNS SD collector supporting multiple record types
+ */
+@Slf4j
+public class DnsSdCollectImpl extends AbstractCollect {
+
+    private static final int DEFAULT_TIME_OUT = 3;
+
+    @Override
+    public void preCheck(Metrics metrics) throws IllegalArgumentException {
+        if (metrics.getDns_sd() == null) {
+            throw new IllegalArgumentException("DNS SD configuration cannot be null");
+        }
+        if (metrics.getDns_sd().getHost() == null || metrics.getDns_sd().getHost().isEmpty()) {
+            throw new IllegalArgumentException("DNS host cannot be null or empty");
+        }
+        if (metrics.getDns_sd().getPort() == null || metrics.getDns_sd().getPort().isEmpty()) {
+            throw new IllegalArgumentException("DNS port cannot be null or empty");
+        }
+        if (metrics.getDns_sd().getRecordType() == null || metrics.getDns_sd().getRecordType().isEmpty()) {
+            throw new IllegalArgumentException("DNS record type cannot be null or empty");
+        }
+        if (metrics.getDns_sd().getRecordName() == null || metrics.getDns_sd().getRecordName().isEmpty()) {
+            throw new IllegalArgumentException("DNS record name cannot be null or empty");
+        }
+    }
+
+    @Override
+    public void collect(CollectRep.MetricsData.Builder builder, Metrics metrics) {
+        String hostName = metrics.getDns_sd().getHost();
+        int type = Integer.parseInt(metrics.getDns_sd().getRecordType());
+        Type.check(type);
+        String recordName = metrics.getDns_sd().getRecordName();
+        try {
+            Lookup lookup = new Lookup(recordName, type);
+            SimpleResolver resolver = new SimpleResolver(metrics.getDns_sd().getHost());
+            resolver.setPort(Integer.parseInt(metrics.getDns_sd().getPort()));
+            resolver.setTimeout(Duration.ofMillis(DEFAULT_TIME_OUT));
+            lookup.setResolver(resolver);
+            lookup.setCache(null);
+            lookup.run();
+            if (lookup.getResult() != Lookup.SUCCESSFUL) {
+                String msg = String.format("DNS lookup failed for: %s, error: %s", recordName, lookup.getErrorString());
+                log.warn(msg);
+                builder.setCode(CollectRep.Code.FAIL);
+                builder.setMsg(msg);
+                return;
+            }
+            Record[] records = lookup.getAnswers();
+            if (records == null || records.length == 0) {
+                log.info("No record type: {} records found for host: {}", type, hostName);
+                builder.setCode(CollectRep.Code.SUCCESS);
+                return;
+            }
+            processRecords(builder, records, type);
+        } catch (TextParseException e) {
+            String errorMsg = CommonUtil.getMessageFromThrowable(e);
+            log.warn("Failed to parse dns query... {}", errorMsg);
+            builder.setCode(CollectRep.Code.FAIL);
+            builder.setMsg(errorMsg);
+        } catch (Exception e) {
+            String errorMsg = CommonUtil.getMessageFromThrowable(e);
+            log.error("Failed to fetch dns sd...{}", errorMsg);
+            builder.setCode(CollectRep.Code.FAIL);
+            builder.setMsg(errorMsg);
+        }
+    }
+
+    private void processRecords(CollectRep.MetricsData.Builder builder, Record[] records, int recordType) {
+        switch (recordType) {
+            case Type.A:
+                processARecords(builder, records);
+                break;
+            case Type.AAAA:
+                processAAAARecords(builder, records);
+                break;
+            case Type.SRV:
+                processSrvRecords(builder, records);
+                break;
+            case Type.MX:
+                processMxRecords(builder, records);
+                break;
+            case Type.NS:
+                processNsRecords(builder, records);
+                break;
+            default:
+                throw new IllegalStateException("Invalid record type: " + recordType);
+        }
+    }
+
+    @SuppressWarnings({"checkstyle:AbbreviationAsWordInName", "checkstyle:LambdaParameterName"})
+    private void processARecords(CollectRep.MetricsData.Builder builder, Record[] records) {
+        Arrays.stream(records).filter(ARecord.class::isInstance).map(ARecord.class::cast).forEach(aRecord -> {
+            CollectRep.ValueRow.Builder row = CollectRep.ValueRow.newBuilder();
+            row.addColumn(aRecord.getAddress().getHostAddress());
+            row.addColumn(""); //A record has no port
+            builder.addValueRow(row.build());
+        });
+    }
+
+    @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
+    private void processAAAARecords(CollectRep.MetricsData.Builder builder, Record[] records) {
+        Arrays.stream(records).filter(AAAARecord.class::isInstance).map(AAAARecord.class::cast).forEach(aaaaRecord -> {
+            CollectRep.ValueRow.Builder row = CollectRep.ValueRow.newBuilder();
+            row.addColumn(aaaaRecord.getAddress().getHostAddress());
+            row.addColumn(""); //AAAA record has no port
+            builder.addValueRow(row.build());
+        });
+    }
+
+    private void processSrvRecords(CollectRep.MetricsData.Builder builder, Record[] records) {
+        Arrays.stream(records).filter(SRVRecord.class::isInstance).map(SRVRecord.class::cast).forEach(srvRecord -> {
+            CollectRep.ValueRow.Builder row = CollectRep.ValueRow.newBuilder();
+            row.addColumn(srvRecord.getTarget().toString(true));
+            row.addColumn(String.valueOf(srvRecord.getPort()));
+            builder.addValueRow(row.build());
+        });
+    }
+
+    private void processMxRecords(CollectRep.MetricsData.Builder builder, Record[] records) {
+        Arrays.stream(records).filter(MXRecord.class::isInstance).map(MXRecord.class::cast).forEach(mxRecord -> {
+            CollectRep.ValueRow.Builder row = CollectRep.ValueRow.newBuilder();
+            row.addColumn(mxRecord.getTarget().toString(true));
+            row.addColumn(""); //MX record has no port
+            builder.addValueRow(row.build());
+        });
+    }
+
+    private void processNsRecords(CollectRep.MetricsData.Builder builder, Record[] records) {
+        Arrays.stream(records).filter(NSRecord.class::isInstance).map(NSRecord.class::cast).forEach(nsRecord -> {
+            CollectRep.ValueRow.Builder row = CollectRep.ValueRow.newBuilder();
+            row.addColumn(nsRecord.getTarget().toString(true));
+            row.addColumn(""); //NS record has no port
+            builder.addValueRow(row.build());
+        });
+    }
+
+    @Override
+    public String supportProtocol() {
+        return DispatchConstants.PROTOCOL_DNS_SD;
+    }
+}
@@ -29,6 +29,7 @@ org.apache.hertzbeat.collector.collect.mqtt.MqttCollectImpl
 org.apache.hertzbeat.collector.collect.ipmi2.IpmiCollectImpl
 org.apache.hertzbeat.collector.collect.kafka.KafkaCollectImpl
 org.apache.hertzbeat.collector.collect.sd.HttpSdCollectImpl
+org.apache.hertzbeat.collector.collect.sd.DnsSdCollectImpl
 org.apache.hertzbeat.collector.collect.sd.EurekaSdCollectImpl
 org.apache.hertzbeat.collector.collect.sd.ConsulSdCollectImpl
 org.apache.hertzbeat.collector.collect.modbus.ModbusCollectImpl
@@ -127,10 +127,15 @@ public interface DispatchConstants {
     * protocol http sd
     */
    String PROTOCOL_HTTP_SD = "http_sd";
+    /**
+     * protocol dns sd
+     */
+    String PROTOCOL_DNS_SD = "dns_sd";
    /**
     * protocol eureka sd
     */
    String PROTOCOL_EUREKA_SD = "eureka_sd";
+
    /**
     * protocol consul sd
     */
@@ -30,9 +30,9 @@ import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.common.entity.job.protocol.DnsProtocol;
+import org.apache.hertzbeat.common.entity.job.protocol.DnsSdProtocol;
 import org.apache.hertzbeat.common.entity.job.protocol.FtpProtocol;
 import org.apache.hertzbeat.common.entity.job.protocol.HttpProtocol;
-import org.apache.hertzbeat.common.entity.job.protocol.HttpSdProtocol;
 import org.apache.hertzbeat.common.entity.job.protocol.EurekaSdProtocol;
 import org.apache.hertzbeat.common.entity.job.protocol.ConsulSdProtocol;
 import org.apache.hertzbeat.common.entity.job.protocol.ModbusProtocol;
@@ -265,7 +265,11 @@ public class Metrics {
    /**
     * http sd protocol
     */
-    private HttpSdProtocol http_sd;
+    private HttpProtocol http_sd;
+    /**
+     * dns sd protocol
+     */
+    private DnsSdProtocol dns_sd;
    /**
     * eureka sd protocol
     */
@@ -0,0 +1,41 @@
+/*
+ * 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 lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Dns protocol
+ */
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class DnsSdProtocol implements Protocol {
+    
+    private String host;
+
+    private String port;
+
+    private String recordType;
+
+    private String recordName;
+}
@@ -77,7 +77,7 @@ public class Monitor {
    @Size(max = 100)
    private String app;
    
-    @Schema(title = "Scrape type: static | http_sd", example = "static", accessMode = READ_WRITE)
+    @Schema(title = "Scrape type: static | http_sd | dns_sd", example = "static", accessMode = READ_WRITE)
    @Size(max = 100)
    private String scrape;
    
@@ -97,6 +97,11 @@ public class ServiceDiscoveryWorker implements InitializingBean {
                            .stream().collect(Collectors.toMap(MonitorBind::getKeyStr, item -> item));
                    RowWrapper rowWrapper = metricsData.readRow();
                    Map<String, String> fieldsValue = Maps.newHashMapWithExpectedSize(8);
+                    String defaultPort = mainMonitorParams.stream()
+                            .filter(param -> FILED_PORT.equals(param.getField()))
+                            .findFirst()
+                            .map(Param::getParamValue)
+                            .orElse("");
                    while (rowWrapper.hasNextRow()) {
                        rowWrapper = rowWrapper.nextRow();
                        fieldsValue.clear();
@@ -105,7 +110,7 @@ public class ServiceDiscoveryWorker implements InitializingBean {
                            fieldsValue.put(cell.getField().getName(), value);
                        });
                        final String host = fieldsValue.get(FILED_HOST);
-                        final String port = fieldsValue.get(FILED_PORT);
+                        final String port = fieldsValue.getOrDefault(FILED_PORT, defaultPort);
                        final String keyStr = host + ":" + port;
                        if (subMonitorBindMap.containsKey(keyStr)) {
                            subMonitorBindMap.remove(keyStr);
@@ -22,6 +22,7 @@ import org.apache.hertzbeat.common.entity.manager.MonitorBind;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
 import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.transaction.annotation.Transactional;

 /**
 * MonitorBind database operation
@@ -33,5 +34,6 @@ public interface MonitorBindDao extends JpaRepository<MonitorBind, Long>, JpaSpe
    void deleteByMonitorId(Long monitorId);
    
    @Modifying
+    @Transactional
    void deleteMonitorBindByBizIdAndMonitorId(Long bizId, Long monitorId);
 }
@@ -0,0 +1,106 @@
+# 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: dns_sd
+# The app api i18n name
+name:
+  zh-CN: Dns Service Discovery
+  en-US: Dns Service Discovery
+# Input params define for app api(render web ui by the definition)
+params:
+  # field-param field key
+  - field: __sd_host__
+    # name-param field display i18n name
+    name:
+      zh-CN: DNS地址
+      en-US: DNS Host
+    # type-param field type(most mapping the html input type)
+    type: text
+    # required-true or false
+    required: true
+  # field-param field key
+  - field: __sd_port__
+    # name-param field display i18n name
+    name:
+      zh-CN: DNS端口
+      en-US: DNS Port
+    # type-param field type(most mapping the html input type)
+    type: text
+    # required-true or false
+    required: true
+    # default value
+    defaultValue: 53
+  - field: __sd_record_type__
+    # name-param field display i18n name
+    name:
+      zh-CN: 记录类型
+      en-US: Record Type
+    # type-param field type(radio mapping the html radio tag)
+    type: radio
+    # required-true or false
+    required: true
+    # when type is radio checkbox, use option to show optional values {name1:value1,name2:value2}
+    options:
+      - label: SRV
+        value: 33
+      - label: A
+        value: 1
+      - label: AAAA
+        value: 28
+      - label: MX
+        value: 15
+      - label: NS
+        value: 2
+  - field: __sd_record_name__
+    # name-param field display i18n name
+    name:
+      zh-CN: 记录名
+      en-US: Record Name
+    # type-param field type(most mapping the html input type)
+    type: text
+    # required-true or false
+    required: true
+metrics:
+  - name: target
+    i18n:
+      zh-CN: 监控目标
+      en-US: Monitor Target
+    # 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
+    # collect metrics content
+    fields:
+      # field-metric name, type-metric type(0-number,1-string), unit-metric unit('%','ms','MB'), label-whether it is a metrics label field
+      - field: host
+        type: 1
+        i18n:
+          zh-CN: Host
+          en-US: Host
+      - field: port
+        type: 1
+        i18n:
+          zh-CN: Port
+          en-US: Port
+    # the protocol used for monitoring, eg: sql, ssh, http, telnet, wmi, snmp, sdk
+    protocol: dns_sd
+    # the config content when protocol is dns_sd
+    dns_sd:
+      host: ^_^__sd_host__^_^
+      port: ^_^__sd_port__^_^
+      recordType: ^_^__sd_record_type__^_^
+      recordName: ^_^__sd_record_name__^_^
@@ -31,6 +31,7 @@
          >
            <nz-option nzValue="static" [nzLabel]="'monitor.scrape.type.static' | i18n"></nz-option>
            <nz-option nzValue="http_sd" [nzLabel]="'monitor.scrape.type.http_sd' | i18n"></nz-option>
+            <nz-option nzValue="dns_sd" [nzLabel]="'monitor.scrape.type.dns_sd' | i18n"></nz-option>
            <nz-option nzValue="eureka_sd" [nzLabel]="'monitor.scrape.type.eureka_sd' | i18n"></nz-option>
            <nz-option nzValue="consul_sd" [nzLabel]="'monitor.scrape.type.consul_sd' | i18n"></nz-option>
          </nz-select>
@@ -762,6 +762,7 @@
  "monitor.sshPrivateKey.tip": "BEGIN RSA PRIVATE KEY",
  "monitor.scrape.type.static": "Static Scrape",
  "monitor.scrape.type.http_sd": "Http Service Discovery",
+  "monitor.scrape.type.dns_sd": "Dns Service Discovery",
  "monitor.scrape.type.eureka_sd": "Eureka Service Discovery",
  "monitor.scrape.type.consul_sd": "Consul Service Discovery",
  "placeholder.key": "Key",
@@ -761,6 +761,7 @@
  "monitor.sshUsername.tip": "SSHトンネルオープン時に必要",
  "monitor.sshPrivateKey.tip": "RSA秘密鍵の起動",
  "monitor.scrape.type.static": "静的な設定",
+  "monitor.scrape.type.dns_sd": "Dns サービスディスカバリー",
  "monitor.scrape.type.http_sd": "Http サービスディスカバリー",
  "monitor.scrape.type.eureka_sd": "Eurekaサービスディスカバリー",
  "monitor.scrape.type.consul_sd": "Consulサービスディスカバリー",
@@ -573,6 +573,7 @@
    "monitor.sshPrivateKey.tip": "Chave privada RSA",
    "monitor.scrape.type.static": "Estatico",
    "monitor.scrape.type.http_sd": "Http Service Discovery",
+    "monitor.scrape.type.dns_sd": "Dns Service Discovery",
    "monitor.scrape.type.eureka_sd": "Eureka Service Discovery",
    "monitor.scrape.type.consul_sd": "Consul Service Discovery",
    "common.name": "Nome da Métrica",
@@ -762,6 +762,7 @@
  "monitor.sshPrivateKey.tip": "启动RSA私钥",
  "monitor.scrape.type.static": "静态配置",
  "monitor.scrape.type.http_sd": "Http 服务发现",
+  "monitor.scrape.type.dns_sd": "Dns 服务发现",
  "monitor.scrape.type.eureka_sd": "Eureka 服务发现",
  "monitor.scrape.type.consul_sd": "Consul 服务发现",
  "placeholder.key": "键",
@@ -761,6 +761,7 @@
  "monitor.sshPrivateKey.tip": "啟動RSA私鑰",
  "monitor.scrape.type.static": "靜態配置",
  "monitor.scrape.type.http_sd": "Http 服務發現",
+  "monitor.scrape.type.dns_sd": "Dns 服務發現",
  "monitor.scrape.type.eureka_sd": "Eureka 服務發現",
  "monitor.scrape.type.consul_sd": "Consul 服務發現",
  "placeholder.key": "鍵",