merge refactoring

add a commons package for templating and Logger interface
core metrics, webhook and memory have their own pakage
move store, models, outside of core

api/assets.go

@@ -8,7 +8,7 @@ import (
 	"github.com/gin-gonic/gin"
 	uuid "github.com/nu7hatch/gouuid"
 	"github.com/vjeantet/bitfan/core"
-	"github.com/vjeantet/bitfan/core/models"
+	"github.com/vjeantet/bitfan/api/models"
 	"github.com/vjeantet/bitfan/entrypoint/parser/logstash"
 )
 

api/client/client.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 
 	"github.com/dghubble/sling"
-	"github.com/vjeantet/bitfan/core/models"
+	"github.com/vjeantet/bitfan/api/models"
 )
 
 type RestClient struct {

api/pipelines.go

@@ -11,7 +11,7 @@ import (
 	"github.com/mitchellh/mapstructure"
 	uuid "github.com/nu7hatch/gouuid"
 	"github.com/vjeantet/bitfan/core"
-	"github.com/vjeantet/bitfan/core/models"
+	"github.com/vjeantet/bitfan/api/models"
 	"github.com/vjeantet/bitfan/entrypoint"
 	"github.com/vjeantet/jodaTime"
 )
@@ -108,16 +108,15 @@ func (p *PipelineApiController) startPipelineByUUID(UUID string) error {
 		return err
 	}
 
-	ppl := loc.ConfigPipeline()
-	ppl.Name = tPipeline.Label
-	ppl.Uuid = tPipeline.Uuid
-
-	agt, err := loc.ConfigAgents()
+	ppl, err := loc.Pipeline()
 	if err != nil {
 		return err
 	}
 
-	nUUID, err := core.StartPipeline(&ppl, agt)
+	ppl.Label = tPipeline.Label
+	ppl.Uuid = tPipeline.Uuid
+
+	nUUID, err := ppl.Start()
 	if err != nil {
 		return err
 	}

api/swagger.json

@@ -35,7 +35,7 @@
           "x-go-name": "Uuid"
         }
       },
-      "x-go-package": "github.com/vjeantet/bitfan/core/models"
+      "x-go-package": "github.com/vjeantet/bitfan/api/models"
     },
     "Error": {
       "description": "A Error is ....\n\nA Error can have.....",
@@ -47,7 +47,7 @@
           "x-go-name": "Message"
         }
       },
-      "x-go-package": "github.com/vjeantet/bitfan/core/models"
+      "x-go-package": "github.com/vjeantet/bitfan/api/models"
     },
     "Pipeline": {
       "description": "A Pipeline is ....\n\nA Pipeline can have.....",
@@ -100,7 +100,7 @@
           "x-go-name": "Uuid"
         }
       },
-      "x-go-package": "github.com/vjeantet/bitfan/core/models"
+      "x-go-package": "github.com/vjeantet/bitfan/api/models"
     },
     "processorDoc": {
       "description": "A Doc is ....\n\nA Doc can have.....",
@@ -198,7 +198,7 @@
           "x-go-name": "Ports"
         }
       },
-      "x-go-package": "github.com/vjeantet/bitfan/core/models"
+      "x-go-package": "github.com/vjeantet/bitfan/api/models"
     }
   }
 }

cmd/bitfan/commands/run.go

@@ -58,8 +58,9 @@ When no configuration is passed to the command, bitfan use the config set in glo
 
 		if !viper.GetBool("no-network") {
 			opt.HttpHandlers = append(opt.HttpHandlers, core.HTTPHandler("/api/v2/", api.Handler("api/v2")))
+
 			if viper.IsSet("prometheus") {
-				opt.HttpHandlers = append(opt.HttpHandlers, core.PrometheusServer(viper.GetString("prometheus.path")))
+				opt.Prometheus = viper.GetString("prometheus.path")
 			}
 		}
 
@@ -116,20 +117,32 @@ When no configuration is passed to the command, bitfan use the config set in glo
 			}
 		}
 
-		for _, loc := range entrypoints.Items {
-			agt, err := loc.ConfigAgents()
-
+		for _, ep := range entrypoints.Items {
+			ppl, err := ep.Pipeline()
 			if err != nil {
-				core.Log().Errorf("Error : %s %v", loc.Path, err)
-				os.Exit(2)
+				core.Log().Fatalln(err)
 			}
-			ppl := loc.ConfigPipeline()
-			_, err = core.StartPipeline(&ppl, agt)
+
+			nUUID, err := ppl.Start()
 			if err != nil {
 				core.Log().Errorf("error: %v", err)
 				os.Exit(1)
 			}
-			core.Log().Infof("Pipeline started %s (%s)", ppl.Name, ppl.Uuid)
+			core.Log().Infof("Pipeline started %s (%s)(%s)", ppl.Label, ppl.Uuid, nUUID)
+
+			// agt, err := ep.ConfigAgents()
+
+			// if err != nil {
+			// 	core.Log().Errorf("Error : %s %v", ep.Path, err)
+			// 	os.Exit(2)
+			// }
+			// ppl := ep.ConfigPipeline()
+			// _, err = core.StartPipeline(&ppl, agt)
+			// if err != nil {
+			// 	core.Log().Errorf("error: %v", err)
+			// 	os.Exit(1)
+			// }
+			// core.Log().Infof("Pipeline started %s (%s)", ppl.Name, ppl.Uuid)
 		}
 
 		if service.Interactive() {

cmd/bitfan/commands/test.go

@@ -6,8 +6,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/vjeantet/bitfan/core"
-	config "github.com/vjeantet/bitfan/core/config"
 	"github.com/vjeantet/bitfan/entrypoint"
 )
 
@@ -58,17 +56,18 @@ var testCmd = &cobra.Command{
 }
 
 func testConfigContent(loc *entrypoint.Entrypoint) error {
-	configAgents, err := loc.ConfigAgents()
-	if err != nil {
-		return err
-	}
+	// TODO : refactor with pipeline
+	// configAgents, err := loc.ConfigAgents()
+	// if err != nil {
+	// 	return err
+	// }
 
-	configAgentsOrdered := config.Sort(configAgents, config.SortInputsFirst)
-	for _, configAgent := range configAgentsOrdered {
-		if _, err := core.NewAgent(configAgent); err != nil {
-			return err
-		}
-	}
+	// configAgentsOrdered := config.Sort(configAgents, config.SortInputsFirst)
+	// for _, configAgent := range configAgentsOrdered {
+	// 	if _, err := core.NewAgent(configAgent); err != nil {
+	// 		return err
+	// 	}
+	// }
 
 	return nil
 }

cmd/bitfanUI/server/server.go

@@ -15,7 +15,7 @@ import (
 	sessions "github.com/tommy351/gin-sessions"
 
 	"github.com/vjeantet/bitfan/api/client"
-	"github.com/vjeantet/bitfan/core/models"
+	"github.com/vjeantet/bitfan/api/models"
 )
 
 var apiClient *client.RestClient

codecs/codec.go

@@ -5,14 +5,14 @@ import (
 	"io"
 
 	"github.com/vjeantet/bitfan/codecs/csv"
-	"github.com/vjeantet/bitfan/codecs/w3c"
 	"github.com/vjeantet/bitfan/codecs/json"
 	"github.com/vjeantet/bitfan/codecs/jsonlines"
-	"github.com/vjeantet/bitfan/codecs/lib"
 	"github.com/vjeantet/bitfan/codecs/line"
 	"github.com/vjeantet/bitfan/codecs/multiline"
 	"github.com/vjeantet/bitfan/codecs/plain"
 	"github.com/vjeantet/bitfan/codecs/rubydebug"
+	"github.com/vjeantet/bitfan/codecs/w3c"
+	"github.com/vjeantet/bitfan/commons"
 	"golang.org/x/text/encoding"
 	"golang.org/x/text/encoding/ianaindex"
 )
@@ -49,11 +49,11 @@ type Codec struct {
 	Role                  string
 	Charset               string
 	Options               map[string]interface{}
-	logger                lib.Logger
+	logger                commons.Logger
 	configWorkingLocation string
 }
 
-func New(name string, conf map[string]interface{}, logger lib.Logger, cwl string) *Codec {
+func New(name string, conf map[string]interface{}, logger commons.Logger, cwl string) *Codec {
 	c := &Codec{
 		Name:                  name,
 		Charset:               "utf-8",

codecs/csv/decoder.go

@@ -8,7 +8,7 @@ import (
 	"io"
 
 	"github.com/mitchellh/mapstructure"
-	"github.com/vjeantet/bitfan/codecs/lib"
+	"github.com/vjeantet/bitfan/commons"
 )
 
 type decoder struct {
@@ -17,7 +17,7 @@ type decoder struct {
 	columnnames []string
 	options     decoderOptions
 	comma       rune
-	log         lib.Logger
+	log         commons.Logger
 	title       bool
 }
 
@@ -68,7 +68,7 @@ func NewDecoder(r io.Reader) *decoder {
 
 	return d
 }
-func (d *decoder) SetOptions(conf map[string]interface{}, logger lib.Logger, cwl string) error {
+func (d *decoder) SetOptions(conf map[string]interface{}, logger commons.Logger, cwl string) error {
 	d.log = logger
 
 	if err := mapstructure.Decode(conf, &d.options); err != nil {

codecs/csv/decoder_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/vjeantet/bitfan/codecs/lib"
+	"github.com/vjeantet/bitfan/commons"
 )
 
 func TestDefaultSettings(t *testing.T) {
@@ -75,7 +75,7 @@ func TestWithCustomColumns(t *testing.T) {
 		"columns":                   []string{"user_defined_1", "user_defined_2"},
 		"comment":                   "#",
 	}
-	var l lib.Logger
+	var l commons.Logger
 	err := d.SetOptions(conf, l, "")
 	assert.NoError(t, err)
 

codecs/decoder.go

@@ -1,9 +1,9 @@
 package codecs
 
-import "github.com/vjeantet/bitfan/codecs/lib"
+import "github.com/vjeantet/bitfan/commons"
 
 type Decoder interface {
 	Decode(*interface{}) error
-	SetOptions(map[string]interface{}, lib.Logger, string) error
+	SetOptions(map[string]interface{}, commons.Logger, string) error
 	More() bool
 }

codecs/encoder.go

@@ -1,8 +1,8 @@
 package codecs
 
-import "github.com/vjeantet/bitfan/codecs/lib"
+import "github.com/vjeantet/bitfan/commons"
 
 type Encoder interface {
 	Encode(map[string]interface{}) error
-	SetOptions(map[string]interface{}, lib.Logger, string) error
+	SetOptions(map[string]interface{}, commons.Logger, string) error
 }

codecs/json/decoder.go

@@ -6,14 +6,14 @@ import (
 	"io"
 
 	"github.com/mitchellh/mapstructure"
-	"github.com/vjeantet/bitfan/codecs/lib"
+	"github.com/vjeantet/bitfan/commons"
 )
 
 type decoder struct {
 	d       *json.Decoder
 	options decoderOptions
 
-	log lib.Logger
+	log commons.Logger
 }
 
 type decoderOptions struct {
@@ -33,7 +33,7 @@ func NewDecoder(r io.Reader) *decoder {
 		options: decoderOptions{},
 	}
 }
-func (d *decoder) SetOptions(conf map[string]interface{}, logger lib.Logger, cwl string) error {
+func (d *decoder) SetOptions(conf map[string]interface{}, logger commons.Logger, cwl string) error {
 	d.log = logger
 
 	err := mapstructure.Decode(conf, &d.options)

codecs/json/encoder.go

@@ -5,7 +5,7 @@ import (
 	"io"
 
 	"github.com/mitchellh/mapstructure"
-	"github.com/vjeantet/bitfan/codecs/lib"
+	"github.com/vjeantet/bitfan/commons"
 )
 
 type encoder struct {
@@ -31,7 +31,7 @@ func NewEncoder(w io.Writer) *encoder {
 	return e
 }
 
-func (e *encoder) SetOptions(conf map[string]interface{}, logger lib.Logger, cwl string) error {
+func (e *encoder) SetOptions(conf map[string]interface{}, logger commons.Logger, cwl string) error {
 	if err := mapstructure.Decode(conf, &e.options); err != nil {
 		return err
 	}

codecs/jsonlines/decoder.go

@@ -8,15 +8,15 @@ import (
 	"strings"
 
 	"github.com/mitchellh/mapstructure"
-	"github.com/vjeantet/bitfan/codecs/lib"
+	"github.com/vjeantet/bitfan/commons"
 )
 
 type decoder struct {
 	more    bool
 	r       *bufio.Scanner
 	options decoderOptions
 
-	log lib.Logger
+	log commons.Logger
 }
 
 type decoderOptions struct {
@@ -63,7 +63,7 @@ func NewDecoder(r io.Reader) *decoder {
 	return d
 }
 
-func (d *decoder) SetOptions(conf map[string]interface{}, logger lib.Logger, cwl string) error {
+func (d *decoder) SetOptions(conf map[string]interface{}, logger commons.Logger, cwl string) error {
 	d.log = logger
 
 	return mapstructure.Decode(conf, &d.options)

codecs/line/decoder.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 
 	"github.com/mitchellh/mapstructure"
-	"github.com/vjeantet/bitfan/codecs/lib"
+	"github.com/vjeantet/bitfan/commons"
 )
 
 // doc decoder
@@ -16,7 +16,7 @@ type decoder struct {
 	r       *bufio.Scanner
 	options decoderOptions
 
-	log lib.Logger
+	log commons.Logger
 }
 
 // doc decoderOptions
@@ -64,7 +64,7 @@ func NewDecoder(r io.Reader) *decoder {
 	return d
 }
 
-func (d *decoder) SetOptions(conf map[string]interface{}, logger lib.Logger, cwl string) error {
+func (d *decoder) SetOptions(conf map[string]interface{}, logger commons.Logger, cwl string) error {
 	d.log = logger
 
 	return mapstructure.Decode(conf, &d.options)