feat: mtail support remote config (#908)

doc/provider.toml

@@ -8,11 +8,12 @@
 #	Version string `json:"version"`
 #
 #	// ConfigMap (InputName -> Config), if version is identical, server side can set Config to nil
-#	Configs map[string]cfg.ConfigWithFormat `json:"configs"`
+#	Configs map[string]map[string]*cfg.ConfigWithFormat `json:"configs"`
 # }
 # type ConfigWithFormat struct {
 # 	Config string       `json:"config"`
 # 	Format ConfigFormat `json:"format"`
+# 	checkSum string `json:"-"`
 # }
 # example response:
 # {
@@ -47,4 +48,4 @@ reload_interval = 120
 # tls_cert = "/etc/categraf/cert.pem"
 # tls_key = "/etc/categraf/key.pem"
 ## Use TLS but skip chain & host verification
-# insecure_skip_verify = false
+# insecure_skip_verify = false

inputs/mtail/internal/mtail/options.go

@@ -23,6 +23,13 @@ type Option interface {
 	apply(*Server) error
 }
 
+type ProgramContent map[string]string
+
+func (opt ProgramContent) apply(m *Server) error {
+	m.rOpts = append(m.rOpts, runtime.ProgramContent(opt))
+	return nil
+}
+
 // ProgramPath sets the path to find mtail programs in the Server.
 type ProgramPath string
 

inputs/mtail/internal/runtime/options.go

@@ -14,6 +14,13 @@ import (
 // Option configures a new program Runtime.
 type Option func(*Runtime) error
 
+func ProgramContent(progs map[string]string) Option {
+	return func(r *Runtime) error {
+		r.progs = progs
+		return nil
+	}
+}
+
 // OverrideLocation sets the timezone location for the VM.
 func OverrideLocation(loc *time.Location) Option {
 	return func(r *Runtime) error {

inputs/mtail/internal/runtime/runtime.go

@@ -10,6 +10,7 @@ import (
 	"bytes"
 	"crypto/sha256"
 	"expvar"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"log"
@@ -48,8 +49,28 @@ const (
 // directory for filesystem changes.  Any compile errors are stored for later retrieival.
 // This function returns an error if an internal error occurs.
 func (r *Runtime) LoadAllPrograms() error {
-	if r.programPath == "" {
-		// glog.V(2).Info("Programpath is empty, loading nothing")
+	if len(r.programPath) == 0 && len(r.progs) == 0 {
+		log.Printf("W! Programpath is empty, loading nothing")
+		return nil
+	}
+	// TODO  load 配置规则
+	if len(r.progs) != 0 {
+		r.programErrorMu.Lock()
+		defer r.programErrorMu.Unlock()
+		for name, prog := range r.progs {
+			if len(strings.TrimSpace(prog)) == 0 {
+				continue
+			}
+			name = fmt.Sprintf("http_%s", name)
+			f := strings.NewReader(prog)
+			r.programErrors[name] = r.CompileAndRun(name, f)
+			if r.programErrors[name] != nil {
+				if r.errorsAbort {
+					return r.programErrors[name]
+				}
+				log.Printf("Compile errors for %s:\n%s", name, r.programErrors[name])
+			}
+		}
 		return nil
 	}
 	s, err := os.Stat(r.programPath)
@@ -105,11 +126,11 @@ func (r *Runtime) LoadAllPrograms() error {
 func (r *Runtime) LoadProgram(programPath string) error {
 	name := filepath.Base(programPath)
 	if strings.HasPrefix(name, ".") {
-		// glog.V(2).Infof("Skipping %s because it is a hidden file.", programPath)
+		log.Printf("W! Skipping %s because it is a hidden file.", programPath)
 		return nil
 	}
 	if filepath.Ext(name) != fileExt {
-		// glog.V(2).Infof("Skipping %s due to file extension.", programPath)
+		log.Printf("W! Skipping %s due to file extension.", programPath)
 		return nil
 	}
 	f, err := os.OpenFile(filepath.Clean(programPath), os.O_RDONLY, 0o600)
@@ -224,6 +245,8 @@ type Runtime struct {
 
 	programPath string // Path that contains mtail programs.
 
+	progs map[string]string
+
 	handleMu sync.RWMutex         // guards accesses to handles
 	handles  map[string]*vmHandle // map of program names to virtual machines
 
@@ -294,7 +317,7 @@ func New(lines <-chan *logline.LogLine, wg *sync.WaitGroup, programPath string,
 		}
 		r.handleMu.Unlock()
 	}()
-	if r.programPath == "" {
+	if len(r.programPath) == 0 && len(r.progs) == 0 {
 		log.Println("No program path specified, no programs will be loaded.")
 		return r, nil
 	}
@@ -304,7 +327,7 @@ func New(lines <-chan *logline.LogLine, wg *sync.WaitGroup, programPath string,
 	go func() {
 		defer r.wg.Done()
 		<-initDone
-		if r.programPath == "" {
+		if len(r.programPath) == 0 && len(r.progs) == 0 {
 			log.Println("no program reload on SIGHUP without programPath")
 			return
 		}

inputs/mtail/mtail.go

@@ -31,6 +31,8 @@ type MTail struct {
 type Instance struct {
 	config.InstanceConfig
 
+	ProgContent map[string]string `toml:"prog_content"`
+
 	NamePrefix           string        `toml:"name_prefix"`
 	Progs                string        `toml:"progs"`
 	Logs                 []string      `toml:"logs"`
@@ -58,7 +60,7 @@ type Instance struct {
 
 func (ins *Instance) Init() error {
 
-	if len(ins.Progs) == 0 || len(ins.Logs) == 0 {
+	if len(ins.Progs) == 0 && len(ins.ProgContent) == 0 || len(ins.Logs) == 0 {
 		return types.ErrInstancesEmpty
 	}
 
@@ -93,6 +95,7 @@ func (ins *Instance) Init() error {
 	}
 
 	opts := []mtail.Option{
+		mtail.ProgramContent(ins.ProgContent),
 		mtail.ProgramPath(ins.Progs),
 		mtail.LogPathPatterns(ins.Logs...),
 		mtail.IgnoreRegexPattern(ins.IgnoreFileRegPattern),