Separate importers from backend.

Separating importers from backend allows to have them running
remotely independently from the backend serving the UI.

An programmatic API is also provided to easily create a new
importer in Go.

Currently, importers are directly connected to the database but
in the long-term the backend should act as a gateway to control
DB accesses.

Author: clems4ever

.travis.yml

@@ -6,3 +6,4 @@ go:
 script:
 - go test -v ./...
 - go build -o go-graphkb cmd/go-graphkb/main.go
+- go build -o source-csv cmd/source-csv/main.go

cmd/go-graphkb/main.go

@@ -4,23 +4,17 @@ import (
 	"context"
 	"fmt"
 	"log"
-	"sort"
-	"strings"
-	"sync"
 	"time"
 
+	"github.com/clems4ever/go-graphkb/internal/database"
 	"github.com/clems4ever/go-graphkb/internal/knowledge"
 	"github.com/clems4ever/go-graphkb/internal/server"
-	"github.com/clems4ever/go-graphkb/internal/sources"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 )
 
-// Sources repository of sources
-var Sources []sources.Source
-
 // Database the selected database
-var Database knowledge.GraphDB
+var Database *database.MariaDB
 
 // ConfigPath string
 var ConfigPath string
@@ -34,12 +28,6 @@ func main() {
 		Use: "go-graphkb [opts]",
 	}
 
-	startCmd := &cobra.Command{
-		Use:  "start",
-		Run:  start,
-		Args: cobra.MaximumNArgs(1),
-	}
-
 	listenCmd := &cobra.Command{
 		Use: "listen",
 		Run: listen,
@@ -61,12 +49,6 @@ func main() {
 		Args: cobra.ExactArgs(1),
 	}
 
-	sourceCmd := &cobra.Command{
-		Use:  "source [source]",
-		Run:  getSource,
-		Args: cobra.ExactArgs(1),
-	}
-
 	queryCmd := &cobra.Command{
 		Use:  "query [query]",
 		Run:  queryFunc,
@@ -77,7 +59,7 @@ func main() {
 
 	cobra.OnInitialize(onInit)
 
-	rootCmd.AddCommand(startCmd, cleanCmd, sourceCmd, listenCmd, countCmd, readCmd, queryCmd)
+	rootCmd.AddCommand(cleanCmd, listenCmd, countCmd, readCmd, queryCmd)
 	if err := rootCmd.Execute(); err != nil {
 		log.Fatal(err)
 	}
@@ -97,75 +79,11 @@ func onInit() {
 	if dbName == "" {
 		log.Fatal("Please provide database_name option in your configuration file")
 	}
-	Database = knowledge.NewMariaDB(
+	Database = database.NewMariaDB(
 		viper.GetString("mariadb.username"),
 		viper.GetString("mariadb.password"),
 		viper.GetString("mariadb.host"),
 		dbName)
-
-	Sources = []sources.Source{
-		sources.NewCSVSource(),
-	}
-
-	for _, s := range Sources {
-		sources.Registry.Add(s)
-		g, err := s.Graph()
-
-		if err != nil {
-			log.Fatal(err)
-		}
-
-		for _, a := range g.Assets() {
-			knowledge.SchemaRegistrySingleton.AddAssetType(a)
-		}
-
-		for _, r := range g.Relations() {
-			knowledge.SchemaRegistrySingleton.AddRelationType(r.Type)
-		}
-	}
-}
-
-func start(cmd *cobra.Command, args []string) {
-	var wg sync.WaitGroup
-	wg.Add(1)
-
-	eventBus := make(chan knowledge.SourceSubGraphUpdates)
-	listener := knowledge.NewSourceListener(Database)
-
-	if err := Database.InitializeSchema(); err != nil {
-		log.Fatal(err)
-	}
-
-	// Start kb listening
-	go listener.Listen(eventBus)
-
-	var selectedSources []sources.Source
-
-	// if argument is provided, we select the source
-	if len(args) == 1 {
-		for _, s := range Sources {
-			if s.Name() == args[0] {
-				selectedSources = []sources.Source{s}
-				break
-			}
-		}
-		if len(selectedSources) == 0 {
-			log.Fatal(fmt.Sprintf("Unable to find source with name %s", args[0]))
-		}
-
-	} else {
-		selectedSources = Sources
-	}
-
-	for _, source := range selectedSources {
-		emitter := knowledge.NewGraphEmitter(source.Name(), eventBus, Database)
-		err := source.Start(emitter)
-		if err != nil {
-			fmt.Printf("[ERROR] %s\n", err)
-		}
-	}
-
-	wg.Wait()
 }
 
 func count(cmd *cobra.Command, args []string) {
@@ -188,53 +106,8 @@ func flush(cmd *cobra.Command, args []string) {
 	fmt.Println("Successul flush")
 }
 
-func getSource(cmd *cobra.Command, args []string) {
-	sourceName := args[0]
-
-	selectedSources := []sources.Source{}
-	for _, s := range Sources {
-		if s.Name() == sourceName {
-			selectedSources = append(selectedSources, s)
-		}
-	}
-
-	assets := make(map[string]bool)
-	relations := make(map[string]bool)
-	for _, s := range selectedSources {
-		g, err := s.Graph()
-		if err != nil {
-			log.Fatal(err)
-		}
-
-		for _, a := range g.Assets() {
-			assets[string(a)] = true
-		}
-
-		for _, r := range g.Relations() {
-			t := fmt.Sprintf("%s_%s_%s", r.FromType, r.Type, r.ToType)
-			relations[t] = true
-		}
-	}
-
-	assetsSlice := make([]string, 0)
-	relationsSlice := make([]string, 0)
-	for a := range assets {
-		assetsSlice = append(assetsSlice, fmt.Sprintf("\t%s", a))
-	}
-	for r := range relations {
-		relationsSlice = append(relationsSlice, fmt.Sprintf("\t%s", r))
-	}
-
-	sort.Strings(assetsSlice)
-	sort.Strings(relationsSlice)
-
-	fmt.Printf("assets -> \n%s\nrelations -> \n%s\n",
-		strings.Join(assetsSlice, "\n"),
-		strings.Join(relationsSlice, "\n"))
-}
-
 func listen(cmd *cobra.Command, args []string) {
-	server.StartServer(Database)
+	server.StartServer(Database, Database)
 }
 
 func read(cmd *cobra.Command, args []string) {

cmd/source-csv/main.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/clems4ever/go-graphkb/importer"
+	"github.com/clems4ever/go-graphkb/internal/sources"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// ConfigPath string
+var ConfigPath string
+
+// SourceName string
+var SourceName string
+
+func onInit() {
+	viper.SetConfigFile(ConfigPath)
+	viper.SetConfigType("yaml")
+
+	if err := viper.ReadInConfig(); err != nil {
+		panic(fmt.Errorf("Cannot read configuration file from %s", ConfigPath))
+	}
+
+	fmt.Println("Using config file:", viper.ConfigFileUsed())
+}
+
+func main() {
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+
+	cobra.OnInitialize(onInit)
+
+	rootCmd := &cobra.Command{
+		Use: "source-csv [opts]",
+		Run: func(cmd *cobra.Command, args []string) {
+			if err := importer.Start(SourceName, sources.NewCSVSource(), nil); err != nil {
+				log.Fatal(err)
+			}
+		},
+	}
+
+	rootCmd.PersistentFlags().StringVar(&ConfigPath, "config", "config.yml", "Provide the path to the configuration file (required)")
+	rootCmd.PersistentFlags().StringVar(&SourceName, "source-name", "", "Provide a unique source name")
+
+	if err := cobra.MarkFlagRequired(rootCmd.PersistentFlags(), "source-name"); err != nil {
+		log.Fatal(err)
+	}
+
+	if err := rootCmd.Execute(); err != nil {
+		log.Fatal(err)
+	}
+}

config.yml

@@ -8,5 +8,4 @@ mariadb:
 
 sources:
   csv:
-    schema: ./examples/example-schema.csv
     data: ./examples/example-data.csv

examples/example-schema.csv

@@ -6,6 +6,7 @@ require (
 	github.com/abbot/go-http-auth v0.4.0
 	github.com/antlr/antlr4 v0.0.0-20200103163232-691acdc23f1f
 	github.com/cheggaaa/pb v2.0.7+incompatible
+	github.com/deckarep/golang-set v1.7.1
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/golang-collections/go-datastructures v0.0.0-20150211160725-59788d5eb259
 	github.com/gorilla/mux v1.7.3

go.mod

@@ -29,6 +29,8 @@ github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwc
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/deckarep/golang-set v1.7.1 h1:SCQV0S6gTtp6itiFrTqI+pfmJ4LN85S1YzhDf9rTHJQ=
+github.com/deckarep/golang-set v1.7.1/go.mod h1:93vsz/8Wt4joVM7c2AVqh+YRMiUSc14yDtF28KmMOgQ=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=

go.sum

@@ -0,0 +1,48 @@
+package importer
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/clems4ever/go-graphkb/internal/database"
+	"github.com/clems4ever/go-graphkb/internal/knowledge"
+	"github.com/clems4ever/go-graphkb/internal/sources"
+	"github.com/spf13/viper"
+)
+
+type ImporterOptions struct {
+	CacheGraph bool
+}
+
+func Start(name string, source sources.Source, options *ImporterOptions) error {
+	dbName := viper.GetString("mariadb.database")
+	if dbName == "" {
+		return fmt.Errorf("Please provide database_name option in your configuration file")
+	}
+	mariaDatabase := database.NewMariaDB(
+		viper.GetString("mariadb.username"),
+		viper.GetString("mariadb.password"),
+		viper.GetString("mariadb.host"),
+		dbName)
+
+	observableSource := sources.NewObservableSource(source)
+
+	eventBus := make(chan knowledge.SourceSubGraphUpdates)
+	listener := knowledge.NewSourceListener(mariaDatabase, mariaDatabase)
+
+	closeC := listener.Listen(eventBus)
+
+	if err := mariaDatabase.InitializeSchema(); err != nil {
+		log.Fatal(err)
+	}
+
+	emitter := knowledge.NewGraphEmitter(name, eventBus, mariaDatabase)
+	if err := observableSource.Start(emitter); err != nil {
+		return err
+	}
+
+	close(eventBus)
+	<-closeC
+
+	return nil
+}