feat: support separate debug & logpath for rules (#2138)

* refactor: use rule options in Topo Signed-off-by: xjasonlyu <[email protected]> * feat: support separate debug & logpath for rules Signed-off-by: xjasonlyu <[email protected]> * docs: add corresponding docs Signed-off-by: xjasonlyu <[email protected]> * test: add rules test options Signed-off-by: xjasonlyu <[email protected]> * fix: recreate logger for specific rule options Signed-off-by: xjasonlyu <[email protected]> * refactor: improve code logic Signed-off-by: xjasonlyu <[email protected]> * refactor: use `logFilename` for rule logs Signed-off-by: xjasonlyu <[email protected]> --------- Signed-off-by: xjasonlyu <[email protected]>
lf-edge · Aug 1, 2023 · cb860de · cb860de
1 parent a0ca9f9
commit cb860de
Show file tree

Hide file tree

Showing 9 changed files with 96 additions and 43 deletions.
diff --git a/docs/en_US/guide/rules/overview.md b/docs/en_US/guide/rules/overview.md
@@ -136,6 +136,8 @@ The current options includes:
 
 | Option name        | Type & Default Value | Description                                                                                                                                                                                                                                                                                                                                       |
 |--------------------|----------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| debug              | bool:false           | Specify whether to enable the debug level for this rule. By default, it will inherit the Debug configuration parameters in the global configuration.                                                                                                                                                                                              |
+| logFilename        | string: ""           | Specify the name of a separate log file for this rule, and the log will be saved in the global log folder. By default, the log configuration parameters in the global configuration will be used.                                                                                                                                                 |
 | isEventTime        | boolean: false       | Whether to use event time or processing time as the timestamp for an event. If event time is used, the timestamp will be extracted from the payload. The timestamp filed must be specified by the [stream](../../sqls/streams.md) definition.                                                                                                     |
 | lateTolerance      | int64:0              | When working with event-time windowing, it can happen that elements arrive late. LateTolerance can specify by how much time(unit is millisecond) elements can be late before they are dropped. By default, the value is 0 which means late elements are dropped.                                                                                  |
 | concurrency        | int: 1               | A rule is processed by several phases of plans according to the sql statement. This option will specify how many instances will be run for each plan. If the value is bigger than 1, the order of the messages may not be retained.                                                                                                               |
@@ -145,9 +147,9 @@ The current options includes:
 | qos                | int:0                | Specify the qos of the stream. The options are 0: At most once; 1: At least once and 2: Exactly once. If qos is bigger than 0, the checkpoint mechanism will be activated to save states periodically so that the rule can be resumed from errors.                                                                                                |
 | checkpointInterval | int:300000           | Specify the time interval in milliseconds to trigger a checkpoint. This is only effective when qos is bigger than 0.                                                                                                                                                                                                                              |
 | restartStrategy    | struct               | Specify the strategy to automatic restarting rule after failures. This can help to get over recoverable failures without manual operations. Please check [Rule Restart Strategy](#rule-restart-strategy) for detail configuration items.                                                                                                          |
-| cron | string: "" | Specify the periodic trigger strategy of the rule, which is described by [cron expression](https://en.wikipedia.org/wiki/Cron) |
-| duration | string: "" | Specifies the running duration of the rule, only valid when cron is specified. The duration should not exceed the time interval between two cron cycles, otherwise it will cause unexpected behavior. |
-| cronDatetimeRange | lists of struct | Specify the effective time period of the Scheduled Rule, which is only valid when `cron` is specified. When this `cronDatetimeRange` is specified, the Scheduled Rule will only take effect within the time range specified. Please see [Scheduled Rule](#Scheduled Rule) for detailed configuration items|
+| cron               | string: ""           | Specify the periodic trigger strategy of the rule, which is described by [cron expression](https://en.wikipedia.org/wiki/Cron)                                                                                                                                                                                                                    |
+| duration           | string: ""           | Specifies the running duration of the rule, only valid when cron is specified. The duration should not exceed the time interval between two cron cycles, otherwise it will cause unexpected behavior.                                                                                                                                             |
+| cronDatetimeRange  | lists of struct      | Specify the effective time period of the Scheduled Rule, which is only valid when `cron` is specified. When this `cronDatetimeRange` is specified, the Scheduled Rule will only take effect within the time range specified. Please see [Scheduled Rule](#Scheduled Rule) for detailed configuration items                                        |
 
 For detail about `qos` and `checkpointInterval`, please check [state and fault tolerance](./state_and_fault_tolerance.md).
 
@@ -177,10 +179,10 @@ When a periodic rule is stopped by [stop rule](../../api/restapi/rules.md#stop-a
 
 `cronDatetimeRange`configuration items are like following:
 
-| Option name  | Type & Default Value | Description                                                                                                                           |
-|--------------|------------|-----------------------------------------------------------|
-| begin     | string    | The begin time of the effective period of the scheduled rule, the format is `YYYY-MM-DD hh:mm:ss'                           |
-| end        | string  | The end time of the effective period of the scheduled rule, the format is `YYYY-MM-DD hh:mm:ss'       |
+| Option name | Type & Default Value | Description                                                                                       |
+|-------------|----------------------|---------------------------------------------------------------------------------------------------|
+| begin       | string               | The begin time of the effective period of the scheduled rule, the format is `YYYY-MM-DD hh:mm:ss' |
+| end         | string               | The end time of the effective period of the scheduled rule, the format is `YYYY-MM-DD hh:mm:ss'   |
 
 `cronDatetimeRange` supports lists of struct, you can declare a set of time ranges to express multiple time ranges for scheduled rules to take effect:
 

diff --git a/docs/zh_CN/guide/rules/overview.md b/docs/zh_CN/guide/rules/overview.md
@@ -138,6 +138,8 @@ eKuiper 已经内置了丰富的 sink connector 类型，如 mqtt、rest 和 fil
 
 | 选项名                | 类型和默认值     | 说明                                                                                             |
 |--------------------|------------|------------------------------------------------------------------------------------------------|
+| debug              | bool:false | 指定该条规则是否开启 Debug Level 的日志水平，缺省情况下会继承全局配置中的 Debug 配置参数。                                        |
+| logFilename        | string: "" | 指定该条规则的单独的日志文件名称，日志将保存在全局日志文件夹中，缺省情况下会延用全局配置中的日志配置参数。                                          |
 | isEventTime        | bool:false | 使用事件时间还是将时间用作事件的时间戳。 如果使用事件时间，则将从有效负载中提取时间戳。 必须通过 [stream](../../sqls/streams.md) 定义指定时间戳记。    |
 | lateTolerance      | int64:0    | 在使用事件时间窗口时，可能会出现元素延迟到达的情况。 LateTolerance 可以指定在删除元素之前可以延迟多少时间（单位为 ms）。 默认情况下，该值为0，表示后期元素将被删除。   |
 | concurrency        | int: 1     | 一条规则运行时会根据 sql 语句分解成多个 plan 运行。该参数设置每个 plan 运行的线程数。该参数值大于1时，消息处理顺序可能无法保证。                      |
@@ -147,9 +149,9 @@ eKuiper 已经内置了丰富的 sink connector 类型，如 mqtt、rest 和 fil
 | qos                | int:0      | 指定流的 qos。 值为0对应最多一次； 1对应至少一次，2对应恰好一次。 如果 qos 大于0，将激活检查点机制以定期保存状态，以便可以从错误中恢复规则。                 |
 | checkpointInterval | int:300000 | 指定触发检查点的时间间隔（单位为 ms）。 仅当 qos 大于0时才有效。                                                          |
 | restartStrategy    | 结构         | 指定规则运行失败后自动重新启动规则的策略。这可以帮助从可恢复的故障中回复，而无需手动操作。请查看[规则重启策略](#规则重启策略)了解详细的配置项目。                    |
-| cron               | string: ""   | 指定规则的周期性触发策略，该周期通过 [cron 表达式](https://zh.wikipedia.org/wiki/Cron) 进行描述。 |
-| duration           | string: ""   | 指定规则的运行持续时间，只有当指定了 cron 后才有效。duration 不应该超过两次 cron 周期之间的时间间隔，否则会引起非预期的行为。   |
-| cronDatetimeRange | 结构体数组 | 指定周期性规则的生效时间段，只有当指定了 cron 后才有效。当指定了该参数后，周期性规则只有在这个参数所制定的时间范围内才生效。请查看 [周期性规则](#周期性规则) 了解详细的配置项目|
+| cron               | string: "" | 指定规则的周期性触发策略，该周期通过 [cron 表达式](https://zh.wikipedia.org/wiki/Cron) 进行描述。                        |
+| duration           | string: "" | 指定规则的运行持续时间，只有当指定了 cron 后才有效。duration 不应该超过两次 cron 周期之间的时间间隔，否则会引起非预期的行为。                      |
+| cronDatetimeRange  | 结构体数组      | 指定周期性规则的生效时间段，只有当指定了 cron 后才有效。当指定了该参数后，周期性规则只有在这个参数所制定的时间范围内才生效。请查看 [周期性规则](#周期性规则) 了解详细的配置项目 |
 
 有关 `qos` 和 `checkpointInterval` 的详细信息，请查看[状态和容错](./state_and_fault_tolerance.md)。
 
@@ -159,13 +161,13 @@ eKuiper 已经内置了丰富的 sink connector 类型，如 mqtt、rest 和 fil
 
 规则重启策略的配置项包括：
 
-| 选项名          | 类型和默认值     | 说明                                                        |
-|--------------|------------|-----------------------------------------------------------|
-| attempts     | int: 0     | 最大重试次数。如果设置为0，该规则将立即失败，不会进行重试。                            |
-| delay        | int: 1000  | 默认的重试间隔时间，以毫秒为单位。如果没有设置 `multiplier`，重试的时间间隔将固定为这个值。       |
+| 选项名          | 类型和默认值     | 说明                                                          |
+|--------------|------------|-------------------------------------------------------------|
+| attempts     | int: 0     | 最大重试次数。如果设置为0，该规则将立即失败，不会进行重试。                              |
+| delay        | int: 1000  | 默认的重试间隔时间，以毫秒为单位。如果没有设置 `multiplier`，重试的时间间隔将固定为这个值。        |
 | maxDelay     | int: 30000 | 重试的最大间隔时间，单位是毫秒。只有当 `multiplier` 有设置时，从而使得每次重试的延迟都会增加时才会生效。 |
-| multiplier   | float: 2   | 重试间隔时间的乘数。                                                |
-| jitterFactor | float: 0.1 | 添加或减去延迟的随机值系数，防止在同一时间重新启动多个规则。                            |
+| multiplier   | float: 2   | 重试间隔时间的乘数。                                                  |
+| jitterFactor | float: 0.1 | 添加或减去延迟的随机值系数，防止在同一时间重新启动多个规则。                              |
 
 这些选项的默认值定义于 `etc/kuiper.yaml` 配置文件，可通过修改该文件更改默认值。
 
@@ -179,10 +181,10 @@ eKuiper 已经内置了丰富的 sink connector 类型，如 mqtt、rest 和 fil
 
 `cronDatetimeRange` 的配置项如下:
 
-| 选项名          | 类型和默认值     | 说明                                                        |
-|--------------|------------|-----------------------------------------------------------|
-| begin     | string    | 周期性规则生效时间段的起始时间，格式为 `YYYY-MM-DD hh:mm:ss"                            |
-| end        | string  | 周期性规则生效时间段的结束时间，格式为 `YYYY-MM-DD hh:mm:ss"       |
+| 选项名   | 类型和默认值 | 说明                                        |
+|-------|--------|-------------------------------------------|
+| begin | string | 周期性规则生效时间段的起始时间，格式为 `YYYY-MM-DD hh:mm:ss" |
+| end   | string | 周期性规则生效时间段的结束时间，格式为 `YYYY-MM-DD hh:mm:ss" |
 
 cronDatetimeRange 支持结构体数组，你可以声明一组时间段来表达周期性规则生效的多个时间段:
 

diff --git a/internal/conf/conf.go b/internal/conf/conf.go
@@ -193,7 +193,7 @@ func SetFileLog(v bool) error {
 		return nil
 	}
 
-	logDir, err := GetLoc(logDir)
+	logDir, err := GetLogLoc()
 	if err != nil {
 		return err
 	}

diff --git a/internal/conf/path.go b/internal/conf/path.go
@@ -55,6 +55,10 @@ func GetConfLoc() (string, error) {
 	return GetLoc(etcDir)
 }
 
+func GetLogLoc() (string, error) {
+	return GetLoc(logDir)
+}
+
 func GetDataLoc() (string, error) {
 	if IsTesting {
 		dataDir, err := GetLoc(dataDir)

diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
@@ -72,7 +72,7 @@ func PlanSQLWithSourcesAndSinks(rule *api.Rule, sources []*node.SourceNode, sink
 
 func createTopo(rule *api.Rule, lp LogicalPlan, sources []*node.SourceNode, sinks []*node.SinkNode, streamsFromStmt []string) (*topo.Topo, error) {
 	// Create topology
-	tp, err := topo.NewWithNameAndQos(rule.Id, rule.Options.Qos, rule.Options.CheckpointInterval)
+	tp, err := topo.NewWithNameAndOptions(rule.Id, rule.Options)
 	if err != nil {
 		return nil, err
 	}

diff --git a/internal/topo/planner/planner_graph.go b/internal/topo/planner/planner_graph.go
@@ -52,7 +52,7 @@ func PlanByGraph(rule *api.Rule) (*topo.Topo, error) {
 	if ruleGraph == nil {
 		return nil, errors.New("no graph")
 	}
-	tp, err := topo.NewWithNameAndQos(rule.Id, rule.Options.Qos, rule.Options.CheckpointInterval)
+	tp, err := topo.NewWithNameAndOptions(rule.Id, rule.Options)
 	if err != nil {
 		return nil, err
 	}

diff --git a/internal/topo/topo.go b/internal/topo/topo.go
@@ -17,8 +17,15 @@ package topo
 import (
 	"context"
 	"fmt"
+	"io"
+	"os"
+	"path"
 	"strconv"
 	"sync"
+	"time"
+
+	rotatelogs "github.com/lestrrat-go/file-rotatelogs"
+	"github.com/sirupsen/logrus"
 
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/checkpoint"
@@ -31,26 +38,24 @@ import (
 )
 
 type Topo struct {
-	sources            []node.DataSourceNode
-	sinks              []*node.SinkNode
-	ctx                api.StreamContext
-	cancel             context.CancelFunc
-	drain              chan error
-	ops                []node.OperatorNode
-	name               string
-	qos                api.Qos
-	checkpointInterval int
-	store              api.Store
-	coordinator        *checkpoint.Coordinator
-	topo               *api.PrintableTopo
-	mu                 sync.Mutex
+	sources     []node.DataSourceNode
+	sinks       []*node.SinkNode
+	ctx         api.StreamContext
+	cancel      context.CancelFunc
+	drain       chan error
+	ops         []node.OperatorNode
+	name        string
+	options     *api.RuleOption
+	store       api.Store
+	coordinator *checkpoint.Coordinator
+	topo        *api.PrintableTopo
+	mu          sync.Mutex
 }
 
-func NewWithNameAndQos(name string, qos api.Qos, checkpointInterval int) (*Topo, error) {
+func NewWithNameAndOptions(name string, options *api.RuleOption) (*Topo, error) {
 	tp := &Topo{
-		name:               name,
-		qos:                qos,
-		checkpointInterval: checkpointInterval,
+		name:    name,
+		options: options,
 		topo: &api.PrintableTopo{
 			Sources: make([]string, 0),
 			Edges:   make(map[string][]interface{}),
@@ -121,6 +126,38 @@ func (s *Topo) addEdge(from api.TopNode, to api.TopNode, toType string) {
 func (s *Topo) prepareContext() {
 	if s.ctx == nil || s.ctx.Err() != nil {
 		contextLogger := conf.Log.WithField("rule", s.name)
+		if s.options.Debug || s.options.LogFilename != "" {
+			contextLogger.Logger = &logrus.Logger{
+				Out:          conf.Log.Out,
+				Hooks:        conf.Log.Hooks,
+				Level:        conf.Log.Level,
+				Formatter:    conf.Log.Formatter,
+				ReportCaller: conf.Log.ReportCaller,
+				ExitFunc:     conf.Log.ExitFunc,
+				BufferPool:   conf.Log.BufferPool,
+			}
+			if conf.Config.Basic.Debug || s.options.Debug {
+				contextLogger.Logger.SetLevel(logrus.DebugLevel)
+			}
+			if s.options.LogFilename != "" {
+				logDir, _ := conf.GetLogLoc()
+
+				file := path.Join(logDir, path.Base(s.options.LogFilename))
+				output, err := rotatelogs.New(
+					file+".%Y-%m-%d_%H-%M-%S",
+					rotatelogs.WithLinkName(file),
+					rotatelogs.WithRotationTime(time.Hour*time.Duration(conf.Config.Basic.RotateTime)),
+					rotatelogs.WithMaxAge(time.Hour*time.Duration(conf.Config.Basic.MaxAge)),
+				)
+				if err != nil {
+					conf.Log.Warnf("Create rule log file failed: %s", file)
+				} else if conf.Config.Basic.ConsoleLog {
+					contextLogger.Logger.SetOutput(io.MultiWriter(output, os.Stdout))
+				} else if !conf.Config.Basic.ConsoleLog {
+					contextLogger.Logger.SetOutput(output)
+				}
+			}
+		}
 		ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
 		s.ctx, s.cancel = ctx.WithCancel()
 	}
@@ -141,7 +178,7 @@ func (s *Topo) Open() <-chan error {
 			s.mu.Lock()
 			defer s.mu.Unlock()
 			var err error
-			if s.store, err = state.CreateStore(s.name, s.qos); err != nil {
+			if s.store, err = state.CreateStore(s.name, s.options.Qos); err != nil {
 				return fmt.Errorf("topo %s create store error %v", s.name, err)
 			}
 			s.enableCheckpoint()
@@ -175,7 +212,7 @@ func (s *Topo) Open() <-chan error {
 }
 
 func (s *Topo) enableCheckpoint() error {
-	if s.qos >= api.AtLeastOnce {
+	if s.options.Qos >= api.AtLeastOnce {
 		var sources []checkpoint.StreamTask
 		for _, r := range s.sources {
 			sources = append(sources, r)
@@ -188,7 +225,7 @@ func (s *Topo) enableCheckpoint() error {
 		for _, r := range s.sinks {
 			sinks = append(sinks, r)
 		}
-		c := checkpoint.NewCoordinator(s.name, sources, ops, sinks, s.qos, s.store, s.checkpointInterval, s.ctx)
+		c := checkpoint.NewCoordinator(s.name, sources, ops, sinks, s.options.Qos, s.store, s.options.CheckpointInterval, s.ctx)
 		s.coordinator = c
 	}
 	return nil

diff --git a/internal/topo/topotest/rule_test.go b/internal/topo/topotest/rule_test.go
@@ -328,6 +328,12 @@ func TestLimitSQL(t *testing.T) {
 			BufferLength: 100,
 			SendError:    true,
 		},
+		{
+			BufferLength: 100,
+			SendError:    true,
+			Debug:        true,
+			LogFilename:  "rule-test.log",
+		},
 		{
 			BufferLength:       100,
 			SendError:          true,