Ver2.0.0 alpha.18 (#26)

* fix: config UnmarshalJSON * fix nats input * update: InputServer err msg
mimuret · Sep 8, 2023 · 9d96e09 · 9d96e09
1 parent 433fae2
commit 9d96e09
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 22 deletions.
diff --git a/pkg/config/config.go b/pkg/config/config.go
@@ -65,20 +65,8 @@ type Config struct {
 	OutputGroups         []OutputGroupConfig
 }
 
-func NewConfig() *Config {
-	return &Config{
-		MetricsListen: ":9520",
-		LogLevel:      "info",
-		InputBufferConfig: &BufferConfig{
-			Name: "input",
-			Size: DefaultInputBufferSize,
-		},
-		InputFilterWorkerNum: DefaultInputFilterWorkerNum,
-	}
-}
-
 func LoadConfig(fs afero.Fs, cfgFile string) (*Config, error) {
-	c := NewConfig()
+	c := &Config{}
 	bs, err := afero.ReadFile(fs, cfgFile)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to open config file")
@@ -127,7 +115,15 @@ func (c *Config) UnmarshalJSON(bs []byte) error {
 		Inputs               json.RawMessage
 		Filters              json.RawMessage
 		OutputGroups         []json.RawMessage
-	}{}
+	}{
+		MetricsListen: ":9520",
+		LogLevel:      "info",
+		InputBufferConfig: &BufferConfig{
+			Name: "input",
+			Size: DefaultInputBufferSize,
+		},
+		InputFilterWorkerNum: DefaultInputFilterWorkerNum,
+	}
 
 	if err := json.Unmarshal(bs, &cfg); err != nil {
 		return errors.Wrap(err, "invalid json Input")
@@ -142,8 +138,10 @@ func (c *Config) UnmarshalJSON(bs []byte) error {
 	if err := json.Unmarshal(cfg.Inputs, &c.Inputs); err != nil {
 		results = gerrors.Join(results, errors.Wrap(err, "failed to create input plugins"))
 	}
-	if err := json.Unmarshal(cfg.Filters, &c.Filters); err != nil {
-		results = gerrors.Join(results, errors.Wrap(err, "failed to create global filter plugins"))
+	if cfg.Filters != nil {
+		if err := json.Unmarshal(cfg.Filters, &c.Filters); err != nil {
+			results = gerrors.Join(results, errors.Wrap(err, "failed to create global filter plugins"))
+		}
 	}
 	for i, v := range cfg.OutputGroups {
 		var og OutputGroupConfig

diff --git a/pkg/plugin/input/input.go b/pkg/plugin/input/input.go
@@ -124,7 +124,7 @@ func (i *InputServer) Serve(ln net.Listener, buf types.Writer) error {
 func (i *InputServer) Read(r io.Reader, buf types.Writer) error {
 	decoder, err := framestream.NewDecoder(r, i.DecoderOptions)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "failed to create fstrm decoder")
 	}
 LOOP:
 	for {

diff --git a/pkg/plugin/input/nats/nats.go b/pkg/plugin/input/nats/nats.go
@@ -143,18 +143,24 @@ func (f *Nats) Subscribe(ctx context.Context, w types.Writer) error {
 	defer func() {
 		_ = sub.Unsubscribe()
 	}()
-
+	wg := sync.WaitGroup{}
+	defer wg.Wait()
 	f.ic.Logger.Info("start subscribe", zap.String("subject", f.Subject), zap.String("queue name", f.QueueName), zap.Int("queue len", f.QueueLen))
 LOOP:
 	for {
 		select {
 		case <-ctx.Done():
 			break LOOP
 		case msg := <-ch:
-			buf := bytes.NewBuffer(msg.Data)
-			if err := is.Read(buf, w); err != nil {
-				return err
-			}
+			wg.Add(1)
+			go func(bs []byte) {
+				buf := bytes.NewBuffer(bs)
+				if err := is.Read(buf, w); err != nil {
+					input.TotalDecordError.Inc()
+					f.ic.Logger.Debug("input error", zap.Error(err))
+				}
+				wg.Done()
+			}(msg.Data)
 		}
 	}
 	return nil