feat: operator 支持 httpsd 服务发现机制 (#515)

pkg/operator/common/compressor/compressor.go → pkg/operator/common/gzip/gzip.go

@@ -7,44 +7,27 @@
 // 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 compressor
+package gzip
 
 import (
 	"bytes"
 	"compress/gzip"
 	"io"
 )
 
-type Compressor interface {
-	Compress(b []byte) ([]byte, error)
-	Uncompress(b []byte) ([]byte, error)
-}
-
 func Compress(b []byte) ([]byte, error) {
-	return defaultCompressor.Compress(b)
-}
-
-func Uncompress(b []byte) ([]byte, error) {
-	return defaultCompressor.Uncompress(b)
-}
-
-var defaultCompressor = gzipCompressor{}
-
-type gzipCompressor struct{}
-
-func (gzipCompressor) Compress(b []byte) ([]byte, error) {
 	buf := &bytes.Buffer{}
 	w := gzip.NewWriter(buf)
 	if _, err := w.Write(b); err != nil {
-		w.Close()
+		_ = w.Close()
 		return nil, err
 	}
-	w.Close()
+	_ = w.Close()
 	return buf.Bytes(), nil
 }
 
-func (gzipCompressor) Uncompress(conf []byte) ([]byte, error) {
-	reader := bytes.NewReader(conf)
+func Uncompress(b []byte) ([]byte, error) {
+	reader := bytes.NewReader(b)
 	r, err := gzip.NewReader(reader)
 	if err != nil {
 		return nil, err

pkg/operator/common/compressor/compressor_test.go → pkg/operator/common/gzip/gzip_test.go

@@ -7,7 +7,7 @@
 // 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 compressor
+package gzip
 
 import (
 	"testing"

pkg/operator/common/k8sutils/client.go

@@ -77,7 +77,7 @@ func NewTkexClient(host string, tlsConfig *rest.TLSClientConfig) (tkexversiond.I
 }
 
 func WaitForNamedCacheSync(ctx context.Context, controllerName string, inf cache.SharedIndexInformer) bool {
-	return operator.WaitForNamedCacheSync(ctx, controllerName, new(logconf.Logger), inf)
+	return operator.WaitForNamedCacheSync(ctx, controllerName, logconf.New(controllerName), inf)
 }
 
 func CreateOrUpdateSecret(ctx context.Context, secretClient clientv1.SecretInterface, desired *corev1.Secret) error {

pkg/operator/common/logconf/hook.go

@@ -19,34 +19,18 @@ import (
 )
 
 const (
-	confStdoutPath    = "log.stdout"
-	confFormatPath    = "log.format"
-	confFileNamePath  = "log.filename"
-	confMaxAgePath    = "log.max_age"
-	confMaxSizePath   = "log.max_size"
-	confMaxBackupPath = "log.max_backup"
-	confLogLevelPath  = "log.level"
+	confLoggerLevelPath = "logger.level"
 )
 
 func initConfig() {
-	viper.SetDefault(confStdoutPath, false)
-	viper.SetDefault(confFormatPath, "logfmt")
-	viper.SetDefault(confFileNamePath, "bkmonitor-operator.log")
-	viper.SetDefault(confMaxAgePath, 3)
-	viper.SetDefault(confMaxSizePath, 512)
-	viper.SetDefault(confMaxBackupPath, 5)
-	viper.SetDefault(confLogLevelPath, "error")
+	viper.SetDefault(confLoggerLevelPath, "info")
 }
 
 func updateConfig() {
 	logger.SetOptions(logger.Options{
-		Stdout:     viper.GetBool(confStdoutPath),
-		Format:     viper.GetString(confFormatPath),
-		Filename:   viper.GetString(confFileNamePath),
-		MaxAge:     viper.GetInt(confMaxAgePath),
-		MaxSize:    viper.GetInt(confMaxSizePath),
-		MaxBackups: viper.GetInt(confMaxBackupPath),
-		Level:      viper.GetString(confLogLevelPath),
+		Stdout: true,
+		Format: "logfmt",
+		Level:  viper.GetString(confLoggerLevelPath),
 	})
 }
 

pkg/operator/common/logconf/logger.go

@@ -9,11 +9,38 @@
 
 package logconf
 
-import "github.com/TencentBlueKing/bkmonitor-datalink/pkg/utils/logger"
+import (
+	"bytes"
 
-type Logger struct{}
+	"github.com/go-kit/log"
+
+	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/utils/logger"
+)
+
+type Logger struct {
+	prefix string
+	w      log.Logger
+}
+
+func New(prefix string) *Logger {
+	l := &Logger{
+		w: log.NewLogfmtLogger(writer{
+			prefix: prefix,
+		}),
+	}
+	return l
+}
 
 func (l *Logger) Log(keyvals ...interface{}) error {
-	logger.Debug(keyvals...)
-	return nil
+	return l.w.Log(keyvals...)
+}
+
+type writer struct {
+	prefix string
+}
+
+func (w writer) Write(b []byte) (int, error) {
+	s := string(bytes.TrimSpace(b))
+	logger.Infof("%s\t%s", w.prefix, s)
+	return len(s), nil
 }

pkg/operator/common/utils/utils.go → pkg/operator/common/stringx/utils.go

@@ -7,7 +7,7 @@
 // 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 utils
+package stringx
 
 import "strings"
 

pkg/operator/common/tasks/tasks.go

@@ -13,15 +13,13 @@ import (
 	"fmt"
 )
 
-const (
-	StatefulSetTaskSecretPrefix = "statefulset-worker"
-	DaemonSetTaskSecretPrefix   = "daemonset-worker"
-	EventTaskSecretPrefix       = "event-worker"
-)
-
 const (
 	LabelTaskType = "taskType"
 
+	PrefixStatefulSetTaskSecret = "statefulset-worker"
+	PrefixDaemonSetTaskSecret   = "daemonset-worker"
+	PrefixEventTaskSecret       = "event-worker"
+
 	TaskTypeDaemonSet   = "daemonset"
 	TaskTypeEvent       = "event"
 	TaskTypeStatefulSet = "statefulset"
@@ -36,15 +34,15 @@ func ValidateTaskType(t string) bool {
 }
 
 func GetDaemonSetTaskSecretName(s string) string {
-	return fmt.Sprintf("%s-%s", DaemonSetTaskSecretPrefix, s)
+	return fmt.Sprintf("%s-%s", PrefixDaemonSetTaskSecret, s)
 }
 
 func GetStatefulSetTaskSecretName(i int) string {
-	return fmt.Sprintf("%s-%d", StatefulSetTaskSecretPrefix, i)
+	return fmt.Sprintf("%s-%d", PrefixStatefulSetTaskSecret, i)
 }
 
 func GetEventTaskSecretName() string {
-	return fmt.Sprintf("%s-0", EventTaskSecretPrefix)
+	return fmt.Sprintf("%s-0", PrefixEventTaskSecret)
 }
 
 func GetTaskLabelSelector(s string) string {

pkg/operator/config/hook.go

@@ -45,8 +45,6 @@ func InitConfig() error {
 		return err
 	}
 
-	fmt.Printf("using config file: %s\n", viper.ConfigFileUsed())
-	fmt.Printf("settings: %+v\n", viper.AllSettings())
 	EventBus.Publish(EventConfigPostParse)
 	return nil
 }

pkg/operator/go.mod

@@ -34,6 +34,7 @@ require (
 	github.com/blang/semver/v4 v4.0.0
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/ghodss/yaml v1.0.0
+	github.com/go-kit/log v0.2.1
 	github.com/valyala/bytebufferpool v1.0.0
 )
 
@@ -74,7 +75,6 @@ require (
 	github.com/fatih/color v1.13.0 // indirect
 	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/garyburd/redigo v1.6.2 // indirect
-	github.com/go-kit/log v0.2.1 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
 	github.com/go-logr/logr v1.2.4 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect

pkg/operator/operator/dataidwatcher/watcher.go

@@ -257,8 +257,7 @@ func (w *dataIDWatcher) deleteDataID(dataID *bkv1beta1.DataID) {
 		return
 	}
 
-	logger.Infof("delete DataID, name=%v, id=%v, labels=%v", dataID.Name, dataID.Spec.DataID, dataID.Labels)
-	Publish()
+	Publish() // 发布信号
 }
 
 func (w *dataIDWatcher) deleteMetricDataID(dataID *bkv1beta1.DataID) {