add daemonset telegraf

Makefile

@@ -28,7 +28,9 @@ image: build
 	sudo docker push $(REGISTRY)/onecloud-operator:$(VERSION)
 
 telegraf-init-image: telegraf-init
-   	sudo docker build -f images/telegraf-init/Dockerfile -t $(REGISTRY)/telegraf-init:$(VERSION) .
+	sudo docker build -f images/telegraf-init/Dockerfile -t $(REGISTRY)/telegraf-init:$(VERSION) .
+	sudo docker push $(REGISTRY)/telegraf-init:$(VERSION)
+
 
 fmt:
 	find . -type f -name "*.go" -not -path "./_output/*" \
@@ -40,3 +42,5 @@ mod:
 	go get $(patsubst %,%@master,$(shell GO111MODULE=on go mod edit -print | sed -n -e 's|.*\(yunion.io/x/[a-z].*\) v.*|\1|p' | grep -v '/onecloud$$'))
 	go mod tidy
 	go mod vendor -v
+
+.PHONY: image telegraf-init-image mod fmt telegraf-init build controller-manager

README.md

@@ -10,4 +10,10 @@ See: [./docs/intro.md](./docs/intro.md)
 $ git clone https://github.com/yunionio/onecloud-operator $GOPATH/src/yunion.io/x/onecloud-operator
 $ cd $GOPATH/src/yunion.io/x/onecloud-operator
 $ make
+```
+
+## telegraf-init
+
+```bash
+VERSION=release-1.5 make telegraf-init-image
 ```

cmd/telegraf-init/main.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"strings"
 
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -35,30 +36,50 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+	var masterAddress string
+	if length := len(node.Status.Conditions); length > 0 {
+		if node.Status.Conditions[length-1].Type == v1.NodeReady &&
+			node.Status.Conditions[length-1].Status == v1.ConditionTrue {
+			for _, addr := range node.Status.Addresses {
+				if addr.Type == v1.NodeInternalIP {
+					masterAddress = addr.Address
+					break
+				}
+			}
+		}
+	}
+
 	if node.Labels[constants.OnecloudEnableHostLabelKey] == "enable" {
 		// host enabled
 		checkTelegrafConfigExist()
 	} else {
 		// host disabled
-		generateTelegrafConfig(nodeName, influxAddr)
+		generateTelegrafConfig(nodeName, masterAddress, influxAddr)
+	}
+	if err := os.MkdirAll(TELEGRAF_DIR, 775); err != nil {
+		log.Fatal(err)
 	}
 }
 
 const TELEGRAF_CONF = "/etc/telegraf/telegraf.conf"
+const TELEGRAF_DIR = "/etc/telegraf/telegraf.d"
 
 func checkTelegrafConfigExist() {
 	if _, err := os.Stat(TELEGRAF_CONF); err != nil {
 		log.Fatalf("stat telegraf config file error %s", err)
 	}
 }
 
-func generateTelegrafConfig(nodeName, influxAddr string) {
+func generateTelegrafConfig(nodeName, nodeIp, influxAddr string) {
 	kwargs := map[string]interface{}{
 		"influxdb": map[string]interface{}{
 			"database": "telegraf",
 			"url":      []string{influxAddr},
 		},
 		"hostname": nodeName,
+		"tags": map[string]string{
+			"host_ip": nodeIp,
+		},
 	}
 	conf := getConfig(kwargs)
 	var mode = os.O_WRONLY | os.O_CREATE | os.O_TRUNC

pkg/apis/onecloud/v1alpha1/defaults.go

@@ -47,6 +47,11 @@ const (
 	DefaultOvnImageTag  = DefaultOvnVersion + "-0"
 
 	DefaultInfluxdbImageVersion = "1.7.7"
+
+	DefaultTelegrafImageName     = "telegraf"
+	DefaultTelegrafImageTag      = "release-1.5"
+	DefaultTelegrafInitImageName = "telegraf-init"
+	DefaultTelegrafInitImageTag  = "release-1.5"
 )
 
 func addDefaultingFuncs(scheme *runtime.Scheme) error {
@@ -150,6 +155,20 @@ func SetDefaults_OnecloudClusterSpec(obj *OnecloudClusterSpec, isEE bool) {
 	)
 	obj.OvnNorth.ImagePullPolicy = corev1.PullIfNotPresent
 
+	// telegraf spec
+	obj.Telegraf.InitContainerImage = getImage(
+		obj.ImageRepository, obj.Telegraf.Repository,
+		DefaultTelegrafInitImageName, "",
+		DefaultTelegrafInitImageTag, "",
+	)
+	SetDefaults_DaemonSetSpec(
+		&obj.Telegraf.DaemonSetSpec,
+		getImage(obj.ImageRepository, obj.Telegraf.Repository,
+			DefaultTelegrafImageName, obj.Telegraf.ImageName,
+			DefaultTelegrafImageTag, obj.Telegraf.Tag,
+		),
+	)
+
 	type stateDeploy struct {
 		obj     *StatefulDeploymentSpec
 		size    string

pkg/apis/onecloud/v1alpha1/types.go

@@ -117,6 +117,8 @@ const (
 	EtcdClientComponentType ComponentType = "etcd-client"
 
 	ItsmComponentType ComponentType = "itsm"
+	// Telegraf is monitor agent component type
+	TelegrafComponentType ComponentType = "telegraf"
 )
 
 // ComponentPhase is the current state of component
@@ -193,6 +195,8 @@ type OnecloudClusterSpec struct {
 	Yunionagent DaemonSetSpec `json:"yunionagent"`
 	// Influxdb holds configuration for influxdb
 	Influxdb StatefulDeploymentSpec `json:"influxdb"`
+	// Telegraf holds configuration for telegraf
+	Telegraf TelegrafSpec `json:"telegraf"`
 	// Monitor holds configuration for monitor service
 	Monitor DeploymentSpec `json:"monitor"`
 	// LoadBalancerEndpoint is upstream loadbalancer virtual ip address or DNS domain
@@ -463,6 +467,11 @@ type HostAgentSpec struct {
 	OvnController ContainerSpec
 }
 
+type TelegrafSpec struct {
+	DaemonSetSpec
+	InitContainerImage string
+}
+
 // ContainerSpec is the container spec of a pod
 type ContainerSpec struct {
 	Image           string               `json:"image"`

pkg/controller/cluster/onecloud_cluster_control.go

@@ -132,6 +132,7 @@ func (occ *defaultClusterControl) updateOnecloudCluster(oc *v1alpha1.OnecloudClu
 		components.CloudmonReportHost(),
 		components.ServiceOperator(),
 		components.Itsm(),
+		components.Telegraf(),
 	}
 	var grp errgroup.Group
 	for _, component := range dependComponents {

pkg/manager/component/component.go

@@ -773,7 +773,7 @@ func (m *ComponentManager) newDaemonSet(
 	cfg *v1alpha1.OnecloudClusterConfig,
 	volHelper *VolumeHelper,
 	spec v1alpha1.DaemonSetSpec, updateStrategy apps.DaemonSetUpdateStrategyType,
-	initContainersFactory func() []corev1.Container,
+	initContainers []corev1.Container,
 	containersFactory func([]corev1.VolumeMount) []corev1.Container,
 ) (*apps.DaemonSet, error) {
 	ns := oc.GetNamespace()
@@ -786,11 +786,6 @@ func (m *ComponentManager) newDaemonSet(
 		updateStrategy = apps.RollingUpdateDaemonSetStrategyType
 	}
 
-	var initContainers []corev1.Container
-	if initContainersFactory != nil {
-		initContainers = initContainersFactory()
-	}
-
 	dsName := controller.NewClusterComponentName(ocName, componentType)
 	appDaemonSet := &apps.DaemonSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -1169,3 +1164,7 @@ func (m *ComponentManager) ServiceOperator() manager.Manager {
 func (m *ComponentManager) Itsm() manager.Manager {
 	return newItsmManager(m)
 }
+
+func (m *ComponentManager) Telegraf() manager.Manager {
+	return newTelegrafManager(m)
+}

pkg/manager/component/host.go

@@ -86,7 +86,7 @@ func (m *hostManager) newHostPrivilegedDaemonSet(
 						},
 						{
 							Name:  "HOST_SYSTEM_SERVICES_OFF",
-							Value: "host-deployer,host_sdnagent",
+							Value: "host-deployer,host_sdnagent,telegraf",
 						},
 						{
 							Name:  "OVN_CONTAINER_IMAGE_TAG",

pkg/manager/component/influxdb.go

@@ -15,6 +15,7 @@
 package component
 
 import (
+	"fmt"
 	"path"
 
 	apps "k8s.io/api/apps/v1"
@@ -148,3 +149,8 @@ func (m *influxdbManager) getDeployment(oc *v1alpha1.OnecloudCluster, cfg *v1alp
 func (m *influxdbManager) getDeploymentStatus(oc *v1alpha1.OnecloudCluster) *v1alpha1.DeploymentStatus {
 	return &oc.Status.Influxdb
 }
+
+func getInfluxDBInternalURL(oc *v1alpha1.OnecloudCluster) string {
+	internalAddress := controller.NewClusterComponentName(oc.GetName(), v1alpha1.InfluxdbComponentType)
+	return fmt.Sprintf("https://%s:%d", internalAddress, constants.InfluxdbPort)
+}

pkg/manager/component/telegraf.go

@@ -0,0 +1,180 @@
+package component
+
+import (
+	apps "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"yunion.io/x/onecloud-operator/pkg/apis/constants"
+
+	"yunion.io/x/onecloud-operator/pkg/apis/onecloud/v1alpha1"
+	"yunion.io/x/onecloud-operator/pkg/manager"
+)
+
+type telegrafManager struct {
+	*ComponentManager
+}
+
+func newTelegrafManager(man *ComponentManager) manager.Manager {
+	return &telegrafManager{ComponentManager: man}
+}
+
+func (m *telegrafManager) Sync(oc *v1alpha1.OnecloudCluster) error {
+	return syncComponent(m, oc, oc.Spec.Influxdb.Disable)
+}
+
+func (m *telegrafManager) getDaemonSet(
+	oc *v1alpha1.OnecloudCluster,
+	cfg *v1alpha1.OnecloudClusterConfig,
+) (*apps.DaemonSet, error) {
+	return m.newTelegrafDaemonSet(v1alpha1.TelegrafComponentType, oc, cfg)
+}
+
+func (m *telegrafManager) newTelegrafDaemonSet(
+	cType v1alpha1.ComponentType,
+	oc *v1alpha1.OnecloudCluster,
+	cfg *v1alpha1.OnecloudClusterConfig,
+) (*apps.DaemonSet, error) {
+	dsSpec := oc.Spec.Telegraf
+	containersF := func(volMounts []corev1.VolumeMount) []corev1.Container {
+		return []corev1.Container{
+			{
+				Name:            cType.String(),
+				Image:           dsSpec.Image, // TODO: set default_image
+				ImagePullPolicy: dsSpec.ImagePullPolicy,
+				Command: []string{
+					"/usr/bin/telegraf",
+					"-config", "/etc/telegraf/telegraf.conf",
+					"-config-directory", "/etc/telegraf/telegraf.d",
+				},
+				VolumeMounts: volMounts,
+			},
+		}
+	}
+	initContainers := func(volMounts []corev1.VolumeMount) []corev1.Container {
+		return []corev1.Container{
+			{
+				Name:            cType.String() + "-init",
+				Image:           dsSpec.InitContainerImage,
+				ImagePullPolicy: dsSpec.ImagePullPolicy,
+				Command:         []string{"/bin/telegraf-init"},
+				VolumeMounts:    volMounts,
+				Env: []corev1.EnvVar{
+					{
+						Name: "NODENAME",
+						ValueFrom: &corev1.EnvVarSource{
+							FieldRef: &corev1.ObjectFieldSelector{
+								FieldPath: "spec.nodeName",
+							},
+						},
+					},
+					{
+						Name:  "INFLUXDB_URL",
+						Value: getInfluxDBInternalURL(oc),
+					},
+				},
+			},
+		}
+	}([]corev1.VolumeMount{{
+		Name:      "etc-telegraf",
+		ReadOnly:  false,
+		MountPath: "/etc/telegraf",
+	}})
+	ds, err := m.newDaemonSet(
+		cType, oc, cfg, NewTelegrafVolume(cType, oc), dsSpec.DaemonSetSpec,
+		"", initContainers, containersF,
+	)
+	ds.Spec.Template.Spec.ServiceAccountName = constants.ServiceAccountOnecloudOperator
+	if err != nil {
+		return nil, err
+	}
+	return ds, nil
+}
+
+func NewTelegrafVolume(
+	cType v1alpha1.ComponentType,
+	oc *v1alpha1.OnecloudCluster,
+) *VolumeHelper {
+	var h = &VolumeHelper{
+		cluster:      oc,
+		component:    cType,
+		volumes:      make([]corev1.Volume, 0),
+		volumeMounts: make([]corev1.VolumeMount, 0),
+	}
+	h.volumeMounts = append(h.volumeMounts, []corev1.VolumeMount{
+		{
+			Name:      "etc-telegraf",
+			ReadOnly:  false,
+			MountPath: "/etc/telegraf",
+		},
+		{
+			Name:      "proc",
+			ReadOnly:  false,
+			MountPath: "/proc",
+		},
+		{
+			Name:      "sys",
+			ReadOnly:  false,
+			MountPath: "/sys",
+		},
+		{
+			Name:      "log",
+			ReadOnly:  false,
+			MountPath: "/var/log/telegraf",
+		},
+		{
+			Name:      "run",
+			ReadOnly:  false,
+			MountPath: "/var/run",
+		},
+	}...)
+
+	var volSrcType = corev1.HostPathDirectoryOrCreate
+	var hostPathDirectory = corev1.HostPathDirectory
+	h.volumes = append(h.volumes, []corev1.Volume{
+		{
+			Name: "etc-telegraf",
+			VolumeSource: corev1.VolumeSource{
+				HostPath: &corev1.HostPathVolumeSource{
+					Path: "/etc/telegraf",
+					Type: &volSrcType,
+				},
+			},
+		},
+		{
+			Name: "proc",
+			VolumeSource: corev1.VolumeSource{
+				HostPath: &corev1.HostPathVolumeSource{
+					Path: "/proc",
+					Type: &hostPathDirectory,
+				},
+			},
+		},
+		{
+			Name: "log",
+			VolumeSource: corev1.VolumeSource{
+				HostPath: &corev1.HostPathVolumeSource{
+					Path: "/var/log/telegraf",
+					Type: &volSrcType,
+				},
+			},
+		},
+		{
+			Name: "run",
+			VolumeSource: corev1.VolumeSource{
+				HostPath: &corev1.HostPathVolumeSource{
+					Path: "/var/run",
+					Type: &hostPathDirectory,
+				},
+			},
+		},
+		{
+			Name: "sys",
+			VolumeSource: corev1.VolumeSource{
+				HostPath: &corev1.HostPathVolumeSource{
+					Path: "/sys",
+					Type: &hostPathDirectory,
+				},
+			},
+		},
+	}...)
+	return h
+}