ekuiper-new/internal/conf/conf.go

// Copyright 2021-2023 EMQ Technologies Co., Ltd.
//
// Licensed 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 conf

import (
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path"
	"time"

	"github.com/lestrrat-go/file-rotatelogs"
	"github.com/sirupsen/logrus"

	"github.com/lf-edge/ekuiper/pkg/api"
	"github.com/lf-edge/ekuiper/pkg/cast"
)

const ConfFileName = "kuiper.yaml"

var (
	Config    *KuiperConf
	IsTesting bool
)

type tlsConf struct {
	Certfile string `yaml:"certfile"`
	Keyfile  string `yaml:"keyfile"`
}

type SinkConf struct {
	MemoryCacheThreshold int    `json:"memoryCacheThreshold" yaml:"memoryCacheThreshold"`
	MaxDiskCache         int    `json:"maxDiskCache" yaml:"maxDiskCache"`
	BufferPageSize       int    `json:"bufferPageSize" yaml:"bufferPageSize"`
	EnableCache          bool   `json:"enableCache" yaml:"enableCache"`
	ResendInterval       int    `json:"resendInterval" yaml:"resendInterval"`
	CleanCacheAtStop     bool   `json:"cleanCacheAtStop" yaml:"cleanCacheAtStop"`
	ResendAlterQueue     bool   `json:"resendAlterQueue" yaml:"resendAlterQueue"`
	ResendPriority       int    `json:"resendPriority" yaml:"resendPriority"`
	ResendIndicatorField string `json:"resendIndicatorField" yaml:"resendIndicatorField"`
}

// Validate the configuration and reset to the default value for invalid values.
func (sc *SinkConf) Validate() error {
	var errs error
	if sc.MemoryCacheThreshold < 0 {
		sc.MemoryCacheThreshold = 1024
		Log.Warnf("memoryCacheThreshold is less than 0, set to 1024")
		errs = errors.Join(errs, errors.New("memoryCacheThreshold:memoryCacheThreshold must be positive"))
	}
	if sc.MaxDiskCache < 0 {
		sc.MaxDiskCache = 1024000
		Log.Warnf("maxDiskCache is less than 0, set to 1024000")
		errs = errors.Join(errs, errors.New("maxDiskCache:maxDiskCache must be positive"))
	}
	if sc.BufferPageSize <= 0 {
		sc.BufferPageSize = 256
		Log.Warnf("bufferPageSize is less than or equal to 0, set to 256")
		errs = errors.Join(errs, errors.New("bufferPageSize:bufferPageSize must be positive"))
	}
	if sc.ResendInterval < 0 {
		sc.ResendInterval = 0
		Log.Warnf("resendInterval is less than 0, set to 0")
		errs = errors.Join(errs, errors.New("resendInterval:resendInterval must be positive"))
	}
	if sc.BufferPageSize > sc.MemoryCacheThreshold {
		sc.MemoryCacheThreshold = sc.BufferPageSize
		Log.Warnf("memoryCacheThreshold is less than bufferPageSize, set to %d", sc.BufferPageSize)
		errs = errors.Join(errs, errors.New("memoryCacheThresholdTooSmall:memoryCacheThreshold must be greater than or equal to bufferPageSize"))
	}
	if sc.MemoryCacheThreshold%sc.BufferPageSize != 0 {
		sc.MemoryCacheThreshold = sc.BufferPageSize * (sc.MemoryCacheThreshold/sc.BufferPageSize + 1)
		Log.Warnf("memoryCacheThreshold is not a multiple of bufferPageSize, set to %d", sc.MemoryCacheThreshold)
		errs = errors.Join(errs, errors.New("memoryCacheThresholdNotMultiple:memoryCacheThreshold must be a multiple of bufferPageSize"))
	}
	if sc.BufferPageSize > sc.MaxDiskCache {
		sc.MaxDiskCache = sc.BufferPageSize
		Log.Warnf("maxDiskCache is less than bufferPageSize, set to %d", sc.BufferPageSize)
		errs = errors.Join(errs, errors.New("maxDiskCacheTooSmall:maxDiskCache must be greater than bufferPageSize"))
	}
	if sc.MaxDiskCache%sc.BufferPageSize != 0 {
		sc.MaxDiskCache = sc.BufferPageSize * (sc.MaxDiskCache/sc.BufferPageSize + 1)
		Log.Warnf("maxDiskCache is not a multiple of bufferPageSize, set to %d", sc.MaxDiskCache)
		errs = errors.Join(errs, errors.New("maxDiskCacheNotMultiple:maxDiskCache must be a multiple of bufferPageSize"))
	}
	if sc.ResendPriority < -1 || sc.ResendPriority > 1 {
		sc.ResendPriority = 0
		Log.Warnf("resendPriority is not in [-1, 1], set to 0")
		errs = errors.Join(errs, errors.New("resendPriority:resendPriority must be -1, 0 or 1"))
	}
	return errs
}

type SourceConf struct {
	HttpServerIp   string   `json:"httpServerIp" yaml:"httpServerIp"`
	HttpServerPort int      `json:"httpServerPort" yaml:"httpServerPort"`
	HttpServerTls  *tlsConf `json:"httpServerTls" yaml:"httpServerTls"`
}

func (sc *SourceConf) Validate() error {
	var errs error
	if sc.HttpServerIp == "" {
		sc.HttpServerIp = "0.0.0.0"
	}
	if sc.HttpServerPort <= 0 || sc.HttpServerPort > 65535 {
		Log.Warnf("invalid source.httpServerPort configuration %d, set to 10081", sc.HttpServerPort)
		errs = errors.Join(errs, errors.New("invalidHttpServerPort:httpServerPort must between 0 and 65535"))
		sc.HttpServerPort = 10081
	}
	return errs
}

type SQLConf struct {
	MaxConnections int `yaml:"maxConnections"`
}

type KuiperConf struct {
	Basic struct {
		Debug          bool     `yaml:"debug"`
		ConsoleLog     bool     `yaml:"consoleLog"`
		FileLog        bool     `yaml:"fileLog"`
		RotateTime     int      `yaml:"rotateTime"`
		MaxAge         int      `yaml:"maxAge"`
		TimeZone       string   `yaml:"timezone"`
		Ip             string   `yaml:"ip"`
		Port           int      `yaml:"port"`
		RestIp         string   `yaml:"restIp"`
		RestPort       int      `yaml:"restPort"`
		RestTls        *tlsConf `yaml:"restTls"`
		Prometheus     bool     `yaml:"prometheus"`
		PrometheusPort int      `yaml:"prometheusPort"`
		PluginHosts    string   `yaml:"pluginHosts"`
		Authentication bool     `yaml:"authentication"`
		IgnoreCase     bool     `yaml:"ignoreCase"`
		SQLConf        *SQLConf `yaml:"sql"`
	}
	Rule   api.RuleOption
	Sink   *SinkConf
	Source *SourceConf
	Store  struct {
		Type         string `yaml:"type"`
		ExtStateType string `yaml:"extStateType"`
		Redis        struct {
			Host               string `yaml:"host"`
			Port               int    `yaml:"port"`
			Password           string `yaml:"password"`
			Timeout            int    `yaml:"timeout"`
			ConnectionSelector string `yaml:"connectionSelector"`
		}
		Sqlite struct {
			Name string `yaml:"name"`
		}
	}
	Portable struct {
		PythonBin   string `yaml:"pythonBin"`
		InitTimeout int    `yaml:"initTimeout"`
	}
}

func SetDebugLevel(v bool) {
	lvl := logrus.InfoLevel
	if v {
		lvl = logrus.DebugLevel
	}
	Log.SetLevel(lvl)
}

func SetConsoleAndFileLog(consoleLog, fileLog bool) error {
	if !fileLog {
		if consoleLog {
			Log.SetOutput(os.Stdout)
		}
		return nil
	}

	logDir, err := GetLogLoc()
	if err != nil {
		return err
	}

	file := path.Join(logDir, logFileName)
	logWriter, err := rotatelogs.New(
		file+".%Y-%m-%d_%H-%M-%S",
		rotatelogs.WithLinkName(file),
		rotatelogs.WithRotationTime(time.Hour*time.Duration(Config.Basic.RotateTime)),
		rotatelogs.WithMaxAge(time.Hour*time.Duration(Config.Basic.MaxAge)),
	)

	if err != nil {
		fmt.Printf("Failed to init log file settings: %v", err)
		Log.Infof("Failed to log to file, using default stderr.")
	} else if consoleLog {
		mw := io.MultiWriter(os.Stdout, logWriter)
		Log.SetOutput(mw)
	} else if !consoleLog {
		Log.SetOutput(logWriter)
	}
	gcOutdatedLog(logDir, time.Hour*time.Duration(Config.Basic.MaxAge))
	return nil
}

func InitConf() {
	cpath, err := GetConfLoc()
	if err != nil {
		panic(err)
	}
	kc := KuiperConf{
		Rule: api.RuleOption{
			LateTol:            1000,
			Concurrency:        1,
			BufferLength:       1024,
			CheckpointInterval: 300000, // 5 minutes
			SendError:          true,
			Restart: &api.RestartStrategy{
				Attempts:     0,
				Delay:        1000,
				Multiplier:   2,
				MaxDelay:     30000,
				JitterFactor: 0.1,
			},
		},
	}

	err = LoadConfigFromPath(path.Join(cpath, ConfFileName), &kc)
	if err != nil {
		Log.Fatal(err)
		panic(err)
	}
	Config = &kc
	if 0 == len(Config.Basic.Ip) {
		Config.Basic.Ip = "0.0.0.0"
	}
	if 0 == len(Config.Basic.RestIp) {
		Config.Basic.RestIp = "0.0.0.0"
	}

	if Config.Basic.Debug {
		SetDebugLevel(true)
	}

	if err := SetConsoleAndFileLog(Config.Basic.ConsoleLog, Config.Basic.FileLog); err != nil {
		log.Fatal(err)
	}

	if Config.Basic.TimeZone != "" {
		if err := cast.SetTimeZone(Config.Basic.TimeZone); err != nil {
			Log.Fatal(err)
		}
	}

	if Config.Store.Type == "redis" && Config.Store.Redis.ConnectionSelector != "" {
		if err := RedisStorageConSelectorApply(Config.Store.Redis.ConnectionSelector, Config); err != nil {
			Log.Fatal(err)
		}
	}
	if Config.Store.ExtStateType == "" {
		Config.Store.ExtStateType = "sqlite"
	}

	if Config.Portable.PythonBin == "" {
		Config.Portable.PythonBin = "python"
	}
	if Config.Portable.InitTimeout <= 0 {
		Config.Portable.InitTimeout = 5000
	}
	if Config.Source == nil {
		Config.Source = &SourceConf{}
	}
	_ = Config.Source.Validate()
	if Config.Sink == nil {
		Config.Sink = &SinkConf{}
	}
	_ = Config.Sink.Validate()

	_ = ValidateRuleOption(&Config.Rule)
}

func ValidateRuleOption(option *api.RuleOption) error {
	var errs error
	if option.CheckpointInterval < 0 {
		option.CheckpointInterval = 0
		Log.Warnf("checkpointInterval is negative, set to 0")
		errs = errors.Join(errs, errors.New("invalidCheckpointInterval:checkpointInterval must be greater than 0"))
	}
	if option.Concurrency < 0 {
		option.Concurrency = 1
		Log.Warnf("concurrency is negative, set to 1")
		errs = errors.Join(errs, errors.New("invalidConcurrency:concurrency must be greater than 0"))
	}
	if option.BufferLength < 0 {
		option.BufferLength = 1024
		Log.Warnf("bufferLength is negative, set to 1024")
		errs = errors.Join(errs, errors.New("invalidBufferLength:bufferLength must be greater than 0"))
	}
	if option.LateTol < 0 {
		option.LateTol = 1000
		Log.Warnf("lateTol is negative, set to 1000")
		errs = errors.Join(errs, errors.New("invalidLateTol:lateTol must be greater than 0"))
	}
	if option.Restart != nil {
		if option.Restart.Multiplier <= 0 {
			option.Restart.Multiplier = 2
			Log.Warnf("restart multiplier is negative, set to 2")
			errs = errors.Join(errs, errors.New("invalidRestartMultiplier:restart multiplier must be greater than 0"))
		}
		if option.Restart.Attempts < 0 {
			option.Restart.Attempts = 0
			Log.Warnf("restart attempts is negative, set to 0")
			errs = errors.Join(errs, errors.New("invalidRestartAttempts:restart attempts must be greater than 0"))
		}
		if option.Restart.Delay <= 0 {
			option.Restart.Delay = 1000
			Log.Warnf("restart delay is negative, set to 1000")
			errs = errors.Join(errs, errors.New("invalidRestartDelay:restart delay must be greater than 0"))
		}
		if option.Restart.MaxDelay <= 0 {
			option.Restart.MaxDelay = option.Restart.Delay
			Log.Warnf("restart maxDelay is negative, set to %d", option.Restart.Delay)
			errs = errors.Join(errs, errors.New("invalidRestartMaxDelay:restart maxDelay must be greater than 0"))
		}
		if option.Restart.JitterFactor <= 0 || option.Restart.JitterFactor >= 1 {
			option.Restart.JitterFactor = 0.1
			Log.Warnf("restart jitterFactor must between 0 and 1, set to 0.1")
			errs = errors.Join(errs, errors.New("invalidRestartJitterFactor:restart jitterFactor must between [0, 1)"))
		}
	}
	return errs
}

func init() {
	InitLogger()
	InitClock()
}

func gcOutdatedLog(filePath string, maxDuration time.Duration) {
	entries, err := os.ReadDir(filePath)
	if err != nil {
		Log.Errorf("gc outdated logs when started failed, err:%v", err)
	}
	now := time.Now()
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}
		if isLogOutdated(entry.Name(), now, maxDuration) {
			err := os.Remove(path.Join(filePath, entry.Name()))
			if err != nil {
				Log.Errorf("remove outdated log %v failed, err:%v", entry.Name(), err)
			}
		}
	}
}

func isLogOutdated(name string, now time.Time, maxDuration time.Duration) bool {
	layout := ".2006-01-02_15-04-05"
	logDateExt := path.Ext(name)
	if t, err := time.Parse(layout, logDateExt); err != nil {
		Log.Errorf("parse log %v datetime failed, err:%v", name, err)
		return false
	} else if int64(now.Sub(t))-int64(maxDuration) > 0 {
		return true
	}
	return false
}