Merge pull request #5 from lyft/datacatalog-cache

Datacatalog cache

Gopkg.toml

@@ -51,6 +51,11 @@ required = [
   source = "https://github.com/lyft/flyteidl"
   version = "^0.14.x"
 
+[[constraint]]
+  name = "github.com/lyft/datacatalog"
+  source = "https://github.com/lyft/datacatalog"
+  version = "^0.1.x"
+
 [[constraint]]
   name = "github.com/lyft/flyteplugins"
   source = "https://github.com/lyft/flyteplugins"

config.yaml

@@ -88,6 +88,10 @@ event:
 admin:
   endpoint: localhost:8089
   insecure: true
+catalog-cache:
+  type: catalog
+  endpoint: datacatalog:8089
+  insecure: true
 errors:
   show-source: true
 logger:

pkg/controller/catalog/catalog_client.go

@@ -3,7 +3,10 @@ package catalog
 import (
 	"context"
 
+	"fmt"
+
 	"github.com/lyft/flyteidl/gen/pb-go/flyteidl/core"
+	"github.com/lyft/flytepropeller/pkg/controller/catalog/datacatalog"
 	"github.com/lyft/flytestdlib/logger"
 	"github.com/lyft/flytestdlib/storage"
 )
@@ -13,16 +16,25 @@ type Client interface {
 	Put(ctx context.Context, task *core.TaskTemplate, execID *core.TaskExecutionIdentifier, inputPath storage.DataReference, outputPath storage.DataReference) error
 }
 
-func NewCatalogClient(store storage.ProtobufStore) Client {
+func NewCatalogClient(ctx context.Context, store storage.ProtobufStore) (Client, error) {
 	catalogConfig := GetConfig()
 
 	var catalogClient Client
-	if catalogConfig.Type == LegacyDiscoveryType {
+	var err error
+	switch catalogConfig.Type {
+	case LegacyDiscoveryType:
 		catalogClient = NewLegacyDiscovery(catalogConfig.Endpoint, store)
-	} else if catalogConfig.Type == NoOpDiscoveryType {
+	case DataCatalogType:
+		catalogClient, err = datacatalog.NewDataCatalog(ctx, catalogConfig.Endpoint, catalogConfig.Insecure, store)
+		if err != nil {
+			return nil, err
+		}
+	case NoOpDiscoveryType, "":
 		catalogClient = NewNoOpDiscovery()
+	default:
+		return nil, fmt.Errorf("No such catalog type available: %v", catalogConfig.Type)
 	}
 
 	logger.Infof(context.Background(), "Created Catalog client, type: %v", catalogConfig.Type)
-	return catalogClient
+	return catalogClient, nil
 }

pkg/controller/catalog/discovery_config.go → pkg/controller/catalog/config.go

@@ -21,11 +21,13 @@ type DiscoveryType = string
 const (
 	NoOpDiscoveryType   DiscoveryType = "noop"
 	LegacyDiscoveryType DiscoveryType = "legacy"
+	DataCatalogType     DiscoveryType = "datacatalog"
 )
 
 type Config struct {
-	Type     DiscoveryType `json:"type" pflag:"\"noop\",Discovery Implementation to use"`
-	Endpoint string        `json:"endpoint" pflag:"\"\", Endpoint for discovery service"`
+	Type     DiscoveryType `json:"type" pflag:"\"noop\", Catalog Implementation to use"`
+	Endpoint string        `json:"endpoint" pflag:"\"\", Endpoint for catalog service"`
+	Insecure bool          `json:"insecure" pflag:"false, Use insecure grpc connection"`
 }
 
 // Gets loaded config for Discovery