ubiquitous-computing-scheduling/agent/internal/service/component/component.go

package component

import (
	"errors"
	"git.inspur.com/sbg-jszt/cfn/cfn-schedule-agent/pkg/utils"
	"strings"

	"git.inspur.com/sbg-jszt/cfn/cfn-schedule-agent/internal/model/component"
	"git.inspur.com/sbg-jszt/cfn/cfn-schedule-agent/pkg/go-sysinfo"
	"git.inspur.com/sbg-jszt/cfn/cfn-schedule-agent/pkg/go-sysinfo/types"
	"git.inspur.com/sbg-jszt/cfn/cfn-schedule-agent/pkg/log"

	telegrafConfig "github.com/influxdata/telegraf/config"
	"github.com/influxdata/telegraf/plugins/outputs/http"
	"github.com/influxdata/telegraf/plugins/outputs/prometheus_client"

	_ "github.com/influxdata/telegraf/plugins/parsers/prometheus"
	_ "github.com/influxdata/telegraf/plugins/serializers/json"
	_ "github.com/influxdata/telegraf/plugins/serializers/prometheus"
	_ "github.com/influxdata/telegraf/plugins/serializers/prometheusremotewrite"

	_ "github.com/influxdata/telegraf/plugins/inputs/cpu"    // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/disk"   // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/diskio" // register plugin
	//_ "github.com/influxdata/telegraf/plugins/inputs/all"
	_ "github.com/influxdata/telegraf/plugins/inputs/http"          // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/kernel"        // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/linux_cpu"     // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/mem"           // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/nats"          // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/net"           // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/netstat"       // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/nstat"         // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/processes"     // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/swap"          // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/sysstat"       // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/system"        // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/systemd_units" // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/win_eventlog"  // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/win_services"  // register plugin
	_ "github.com/influxdata/telegraf/plugins/inputs/wireless"      // register plugin
)

func GetComInfo() ([]component.Component, error) {

	var telegrafCom component.Component
	info, err := GetTelegrafProcessInfo()
	if err != nil {
		telegrafCom.COMName = "telegraf"
		telegrafCom.COMPort = 0
		telegrafCom.COMType = "telegraf"
		telegrafCom.COMVersion = "1.28.2"
		telegrafCom.Status = "0"
	} else {
		telegrafCom.COMName = "telegraf"
		telegrafCom.COMPort = 0
		telegrafCom.COMType = "telegraf"
		telegrafCom.COMVersion = "1.28.2"
		telegrafCom.LastStarttime = info.StartTime
		telegrafCom.Status = "1"
	}

	runDirectory, _ := utils.GetCurrentPath()
	conf := runDirectory + "/data-collector/telegraf/conf/telegraf.conf"

	c := telegrafConfig.NewConfig()
	err = c.LoadConfig(conf)
	if err != nil {
		log.Errorf("读取Telegraf配置失败：%s", err)
		log.Errorf("读取Telegraf配置失败：%s", conf)

	} else {
		inputs := c.Inputs
		for _, input := range inputs {
			telegrafCom.Config.Metrics = append(telegrafCom.Config.Metrics, input.Config.Name)
		}

		outputs := c.Outputs
		for _, output := range outputs {

			prometheusClient, ok := output.Output.(*prometheus_client.PrometheusClient)

			if ok {
				ms1 := component.MetricServer{
					Name:    output.Config.Name,
					Address: prometheusClient.Listen + prometheusClient.Path,
				}
				telegrafCom.Config.MetricServer = append(telegrafCom.Config.MetricServer, ms1)
			} else {
				httpClient, ok := output.Output.(*http.HTTP)
				if ok {
					ms2 := component.MetricServer{
						Name:    output.Config.Name,
						Address: httpClient.URL,
					}
					telegrafCom.Config.MetricServer = append(telegrafCom.Config.MetricServer, ms2)
				}
			}

		}

		telegrafCom.Config.Tags = c.Tags
	}

	data := make([]component.Component, 1)
	data[0] = telegrafCom
	return data, nil
}

func GetTelegrafProcessInfo() (*types.ProcessInfo, error) {
	processes, err := sysinfo.Processes()
	if err != nil {
		log.Errorf("心跳检测服务调用失败！")
		return nil, err
	}

	for _, process := range processes {
		info, err := process.Info()

		if err != nil {
			continue
		}

		if strings.Contains(info.Name, "telegraf") || strings.Contains(info.Name, "Telegraf") {
			return &info, nil
		}
	}

	return nil, errors.New("telegraf采集器进程未运行！")
}

func GetProcessExporterProcessInfo() (*types.ProcessInfo, error) {
	processes, err := sysinfo.Processes()
	if err != nil {
		log.Errorf("心跳检测服务调用失败！")
		return nil, err
	}

	for _, process := range processes {
		info, err := process.Info()

		if err != nil {
			continue
		}

		if strings.Contains(info.Name, "process-exporter") || strings.Contains(info.Name, "process-exporter") {
			return &info, nil
		}
	}

	return nil, errors.New("process-exporte采集器进程未运行！")
}