add ntp plugin

README.md

@@ -18,12 +18,14 @@ tar zcvf categraf.tar.gz categraf conf
 
 ## todo
 
-- []ntp
+- [x]ntp
 - []procstat
-- []promscrape
-- []mysql
-- []redis
 - []nginx vts
 - []tomcat
 - []...
+- []promscrape
+- []mysql
+- []redis
+- []statsd
+- []...
 - []io.util

agent/agent.go

@@ -25,6 +25,7 @@ import (
 	_ "flashcat.cloud/categraf/inputs/net"
 	_ "flashcat.cloud/categraf/inputs/netresponse"
 	_ "flashcat.cloud/categraf/inputs/netstat"
+	_ "flashcat.cloud/categraf/inputs/ntp"
 	_ "flashcat.cloud/categraf/inputs/oracle"
 	_ "flashcat.cloud/categraf/inputs/ping"
 	_ "flashcat.cloud/categraf/inputs/processes"

conf/input.ntp/ntp.toml

@@ -0,0 +1,5 @@
+# # collect interval
+# interval = 15
+
+# # ntp servers
+# ntp_servers = ["ntp1.aliyun.com"]

inputs/ntp/ntp.go

@@ -0,0 +1,70 @@
+package ntp
+
+import (
+	"errors"
+	"log"
+	"time"
+
+	"flashcat.cloud/categraf/config"
+	"flashcat.cloud/categraf/inputs"
+	"flashcat.cloud/categraf/types"
+	"github.com/toolkits/pkg/nux"
+)
+
+const inputName = "ntp"
+
+type NTPStat struct {
+	Interval   config.Duration `toml:"interval"`
+	NTPServers []string        `toml:"ntp_servers"`
+	server     string
+}
+
+func init() {
+	inputs.Add(inputName, func() inputs.Input {
+		return &NTPStat{}
+	})
+}
+
+func (n *NTPStat) GetInputName() string {
+	return inputName
+}
+
+func (n *NTPStat) GetInterval() config.Duration {
+	return n.Interval
+}
+
+func (n *NTPStat) Drop() {}
+
+func (n *NTPStat) Init() error {
+	if len(n.NTPServers) == 0 {
+		return errors.New("ntp servers empty")
+	}
+	return nil
+}
+
+func (n *NTPStat) Gather() (samples []*types.Sample) {
+	for _, server := range n.NTPServers {
+		if n.server == "" {
+			n.server = server
+		}
+
+		orgTime := time.Now()
+		serverReciveTime, serverTransmitTime, err := nux.NtpTwoTime(n.server)
+		if err != nil {
+			log.Println("E! failed to connect ntp server:", n.server, "error:", err)
+			n.server = ""
+			continue
+		}
+
+		dstTime := time.Now()
+
+		// https://en.wikipedia.org/wiki/Network_Time_Protocol
+		duration := ((serverReciveTime.UnixNano() - orgTime.UnixNano()) + (serverTransmitTime.UnixNano() - dstTime.UnixNano())) / 2
+
+		delta := duration / 1e6 // convert to ms
+		samples = append(samples, inputs.NewSample("offset_ms", delta))
+		break
+	}
+
+	return
+}