Importers communicate with GraphKB backend over http with authentication.

Importers do not connect directly to DB anymore but instead send their
updates over http with authentication.

Authentication is handled with tokens whitelisted in the backend to allow
an importer to inject data into the graph.

Author: clems4ever

.travis.yml

@@ -5,5 +5,6 @@ go:
 
 script:
 - go test -v ./...
+- pushd web && npm ci && npm run build && popd
 - go build -o go-graphkb cmd/go-graphkb/main.go
 - go build -o source-csv cmd/source-csv/main.go

config.yml renamed to cmd/go-graphkb/config.yml

@@ -7,5 +7,4 @@ mariadb:
   database: graphkb
 
 sources:
-  csv:
-    data: ./examples/example-data.csv
+  csv-employees: token_auth_for_csv_importer

cmd/go-graphkb/main.go

@@ -107,7 +107,18 @@ func flush(cmd *cobra.Command, args []string) {
 }
 
 func listen(cmd *cobra.Command, args []string) {
-	server.StartServer(Database, Database)
+	eventBus := make(chan knowledge.SourceSubGraphUpdates)
+	listener := knowledge.NewGraphUpdater(Database, Database)
+
+	listener.Listen(eventBus)
+
+	if err := Database.InitializeSchema(); err != nil {
+		log.Fatal(err)
+	}
+
+	server.StartServer(Database, Database, eventBus)
+
+	close(eventBus)
 }
 
 func read(cmd *cobra.Command, args []string) {

cmd/importer-csv/config.yml

@@ -0,0 +1,5 @@
+graphkb:
+  url: "http://localhost:8090"
+  auth_token: "token_auth_for_csv_importer"
+
+path: "../../examples/example-data.csv"

cmd/source-csv/main.go renamed to cmd/importer-csv/main.go

@@ -13,18 +13,13 @@ import (
 // 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() {
@@ -35,18 +30,14 @@ func main() {
 	rootCmd := &cobra.Command{
 		Use: "source-csv [opts]",
 		Run: func(cmd *cobra.Command, args []string) {
-			if err := importer.Start(SourceName, sources.NewCSVSource(), nil); err != nil {
+			if err := importer.Start(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)
-	}
+	rootCmd.PersistentFlags().StringVar(&ConfigPath, "config", "config.yml",
+		"Provide the path to the configuration file (required)")
 
 	if err := rootCmd.Execute(); err != nil {
 		log.Fatal(err)

importer/importer.go

@@ -2,9 +2,6 @@ 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"
@@ -14,35 +11,24 @@ 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")
+func Start(source sources.Source, options *ImporterOptions) error {
+	url := viper.GetString("graphkb.url")
+	if url == "" {
+		return fmt.Errorf("Please provide graphkb URL in 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)
+	authToken := viper.GetString("graphkb.auth_token")
+	if authToken == "" {
+		return fmt.Errorf("Please provide a graphkb auth token to communicate with GraphKB")
 	}
 
-	emitter := knowledge.NewGraphEmitter(name, eventBus, mariaDatabase)
-	if err := observableSource.Start(emitter); err != nil {
-		return err
-	}
+	observableSource := sources.NewObservableSource(source)
+	api := knowledge.NewGraphAPI(url, authToken)
+	graphImporter := knowledge.NewGraphImporter(api)
 
-	close(eventBus)
-	<-closeC
+	if err := observableSource.Start(graphImporter); err != nil {
+		return fmt.Errorf("Unable to start importer: %v", err)
+	}
 
 	return nil
 }

internal/database/mariadb.go

@@ -444,7 +444,7 @@ func (m *MariaDB) Query(ctx context.Context, query *query.QueryIL) (*knowledge.G
 }
 
 func (m *MariaDB) SaveSchema(ctx context.Context, sourceName string, schema schema.SchemaGraph) error {
-	b, err := schema.ToJSON()
+	b, err := json.Marshal(schema)
 	if err != nil {
 		return fmt.Errorf("Unable to json encode schema: %v", err)
 	}
@@ -470,7 +470,8 @@ func (m *MariaDB) LoadSchema(ctx context.Context, sourceName string) (schema.Sch
 	}
 
 	graph := schema.NewSchemaGraph()
-	if err := graph.FromJSON([]byte(rawJson)); err != nil {
+	err := json.Unmarshal([]byte(rawJson), &graph)
+	if err != nil {
 		return schema.NewSchemaGraph(), err
 	}