Add support for plugins with dynamic schema. Closes #1012

cmd/plugin.go

@@ -445,7 +445,7 @@ func refreshConnectionsIfNecessary(reports []display.InstallReport, isUpdate boo
 		if err != nil {
 			return err
 		}
-		steampipeconfig.Config = config
+		steampipeconfig.GlobalConfig = config
 	}
 
 	client, err := db_local.GetLocalClient(constants.InvokerPlugin)

cmd/root.go

@@ -104,10 +104,10 @@ func initGlobalConfig() {
 	config, err := steampipeconfig.LoadSteampipeConfig(workspace, cmd.Name())
 	utils.FailOnError(err)
 
-	steampipeconfig.Config = config
+	steampipeconfig.GlobalConfig = config
 
 	// set viper config defaults from config and env vars
-	cmdconfig.SetViperDefaults(steampipeconfig.Config)
+	cmdconfig.SetViperDefaults(steampipeconfig.GlobalConfig)
 }
 
 // CreateLogger :: create a hclog logger with the level specified by the SP_LOG env var

db/db_client/db_client.go

@@ -163,10 +163,10 @@ func (c *DbClient) getSchemaFromDB() (*sql.Rows, error) {
 }
 
 // RefreshConnectionAndSearchPaths implements Client
-func (c *DbClient) RefreshConnectionAndSearchPaths() *db_common.RefreshConnectionResult {
+func (c *DbClient) RefreshConnectionAndSearchPaths() *steampipeconfig.RefreshConnectionResult {
 	// base db client does not refresh connections, it just sets search path
 	// (only local db client refreshed connections)
-	res := &db_common.RefreshConnectionResult{}
+	res := &steampipeconfig.RefreshConnectionResult{}
 	if err := c.SetSessionSearchPath(); err != nil {
 		res.Error = err
 	}

db/db_common/client.go

@@ -32,5 +32,5 @@ type Client interface {
 
 	// remote client will have empty implementation
 
-	RefreshConnectionAndSearchPaths() *RefreshConnectionResult
+	RefreshConnectionAndSearchPaths() *steampipeconfig.RefreshConnectionResult
 }

db/db_local/local_db_client.go

@@ -6,7 +6,6 @@ import (
 	"log"
 	"sort"
 	"strings"
-	"time"
 
 	"github.com/spf13/viper"
 	"github.com/turbot/steampipe/constants"
@@ -137,7 +136,7 @@ func (c *LocalDbClient) SetSessionSearchPath(currentSearchPath ...string) error
 
 // local only functions
 
-func (c *LocalDbClient) RefreshConnectionAndSearchPaths() *db_common.RefreshConnectionResult {
+func (c *LocalDbClient) RefreshConnectionAndSearchPaths() *steampipeconfig.RefreshConnectionResult {
 	res := c.refreshConnections()
 	if res.Error != nil {
 		return res
@@ -239,234 +238,3 @@ func (c *LocalDbClient) getDefaultSearchPath() []string {
 
 	return searchPath
 }
-
-// RefreshConnections loads required connections from config
-// and update the database schema and search path to reflect the required connections
-// return whether any changes have been made
-func (c *LocalDbClient) refreshConnections() *db_common.RefreshConnectionResult {
-
-	res := &db_common.RefreshConnectionResult{}
-	utils.LogTime("db.refreshConnections start")
-	defer utils.LogTime("db.refreshConnections end")
-
-	// load required connection from global config
-	requiredConnections := steampipeconfig.Config.Connections
-
-	// first get a list of all existing schemas
-	schemas := c.client.SchemaMetadata().GetSchemas()
-
-	// refresh the connection state file - the removes any connections which do not exist in the list of current schema
-	updates, err := steampipeconfig.GetConnectionsToUpdate(schemas, requiredConnections)
-	if err != nil {
-		res.Error = err
-		return res
-	}
-	log.Printf("[TRACE] refreshConnections, updates: %+v\n", updates)
-
-	missingCount := len(updates.MissingPlugins)
-	if missingCount > 0 {
-		// if any plugins are missing, error for now but we could prompt for an install
-		res.Error = fmt.Errorf("%d %s referenced in the connection config not installed: \n  %v",
-			missingCount,
-			utils.Pluralize("plugin", missingCount),
-			strings.Join(updates.MissingPlugins, "\n  "))
-		return res
-	}
-
-	var connectionQueries []string
-	numUpdates := len(updates.Update)
-	log.Printf("[TRACE] RefreshConnections: num updates %d", numUpdates)
-	if numUpdates > 0 {
-		// first instantiate connection plugins for all updates (reuse 'res' defined above)
-		var connectionPlugins []*steampipeconfig.ConnectionPlugin
-		connectionPlugins, res = getConnectionPlugins(updates.Update)
-		if res.Error != nil {
-			return res
-		}
-
-		// find any plugins which use a newer sdk version than steampipe.
-		validationFailures, validatedUpdates, validatedPlugins := steampipeconfig.ValidatePlugins(updates.Update, connectionPlugins)
-		if len(validationFailures) > 0 {
-			res.Warnings = append(res.Warnings, steampipeconfig.BuildValidationWarningString(validationFailures))
-		}
-
-		// get schema queries - this updates schemas for validated plugins and drops schemas for unvalidated plugins
-		connectionQueries = getSchemaQueries(validatedUpdates, validationFailures)
-		// add comments queries for validated connections
-		connectionQueries = append(connectionQueries, getCommentQueries(validatedPlugins)...)
-	}
-
-	for c := range updates.Delete {
-		log.Printf("[TRACE] delete connection %s\n ", c)
-		connectionQueries = append(connectionQueries, deleteConnectionQuery(c)...)
-	}
-
-	if len(connectionQueries) == 0 {
-		return res
-	}
-
-	// execute the connection queries
-	if err = executeConnectionQueries(connectionQueries, updates); err != nil {
-		res.Error = err
-		return res
-	}
-
-	// so there ARE connections to update
-
-	// reload the database schemas, since they have changed - otherwise we wouldn't be here
-	log.Println("[TRACE] RefreshConnections: reloading schema")
-	c.LoadSchema()
-
-	res.UpdatedConnections = true
-	return res
-
-}
-
-func (c *LocalDbClient) updateConnectionMap() error {
-	// load the connection state and cache it!
-	log.Println("[TRACE]", "retrieving connection map")
-	connectionMap, err := steampipeconfig.GetConnectionState(c.client.SchemaMetadata().GetSchemas())
-	if err != nil {
-		return err
-	}
-	log.Println("[TRACE]", "setting connection map")
-	c.connectionMap = &connectionMap
-
-	return nil
-}
-
-func getConnectionPlugins(updates steampipeconfig.ConnectionDataMap) ([]*steampipeconfig.ConnectionPlugin, *db_common.RefreshConnectionResult) {
-	res := &db_common.RefreshConnectionResult{}
-	var connectionPlugins []*steampipeconfig.ConnectionPlugin
-
-	// create channels buffered to hold all updates
-	numUpdates := len(updates)
-	var pluginChan = make(chan *steampipeconfig.ConnectionPlugin, numUpdates)
-	var errorChan = make(chan error, numUpdates)
-
-	for connectionName, connectionData := range updates {
-		// instantiate the connection plugin, and retrieve schema
-		go getConnectionPluginAsync(connectionName, connectionData, pluginChan, errorChan)
-	}
-
-	for i := 0; i < numUpdates; i++ {
-		select {
-		case err := <-errorChan:
-			log.Println("[TRACE] get connections err chan select - adding warning", "error", err)
-			res.Warnings = append(res.Warnings, err.Error())
-		case p := <-pluginChan:
-			connectionPlugins = append(connectionPlugins, p)
-		case <-time.After(10 * time.Second):
-			res.Error = fmt.Errorf("timed out retrieving schema from plugins")
-			return nil, res
-		}
-	}
-	return connectionPlugins, res
-}
-
-func getConnectionPluginAsync(connectionName string, connectionData *steampipeconfig.ConnectionData, pluginChan chan *steampipeconfig.ConnectionPlugin, errorChan chan error) {
-	p, err := steampipeconfig.CreateConnectionPlugin(connectionData.Connection, true)
-	if err != nil {
-		errorChan <- err
-		return
-	}
-	pluginChan <- p
-
-	p.Plugin.Client.Kill()
-}
-
-func getSchemaQueries(updates steampipeconfig.ConnectionDataMap, failures []*steampipeconfig.ValidationFailure) []string {
-	var schemaQueries []string
-	for connectionName, plugin := range updates {
-		remoteSchema := steampipeconfig.PluginFQNToSchemaName(plugin.Plugin)
-		log.Printf("[TRACE] update connection %s, plugin Name %s, schema %s\n ", connectionName, plugin.Plugin, remoteSchema)
-		schemaQueries = append(schemaQueries, updateConnectionQuery(connectionName, remoteSchema)...)
-	}
-	for _, failure := range failures {
-		log.Printf("[TRACE] remove schema for conneciton failing validation connection %s, plugin Name %s\n ", failure.ConnectionName, failure.Plugin)
-		if failure.ShouldDropIfExists {
-			schemaQueries = append(schemaQueries, deleteConnectionQuery(failure.ConnectionName)...)
-		}
-	}
-
-	return schemaQueries
-}
-
-func getCommentQueries(plugins []*steampipeconfig.ConnectionPlugin) []string {
-	var commentQueries []string
-	for _, plugin := range plugins {
-		commentQueries = append(commentQueries, commentsQuery(plugin)...)
-	}
-	return commentQueries
-}
-
-func updateConnectionQuery(localSchema, remoteSchema string) []string {
-	// escape the name
-	localSchema = db_common.PgEscapeName(localSchema)
-	return []string{
-
-		// Each connection has a unique schema. The schema, and all objects inside it,
-		// are owned by the root user.
-		fmt.Sprintf(`drop schema if exists %s cascade;`, localSchema),
-		fmt.Sprintf(`create schema %s;`, localSchema),
-		fmt.Sprintf(`comment on schema %s is 'steampipe plugin: %s';`, localSchema, remoteSchema),
-
-		// Steampipe users are allowed to use the new schema
-		fmt.Sprintf(`grant usage on schema %s to steampipe_users;`, localSchema),
-
-		// Permissions are limited to select only, and should be granted for all new
-		// objects. Steampipe users cannot create tables or modify data in the
-		// connection schema - they need to use the public schema for that.  These
-		// commands alter the defaults for any objects created in the future.
-		// See https://www.postgresql.org/docs/12/ddl-priv.html
-		fmt.Sprintf(`alter default privileges in schema %s grant select on tables to steampipe_users;`, localSchema),
-
-		// If there are any objects already then grant their permissions now. (This
-		// should not actually do anything at this point.)
-		fmt.Sprintf(`grant select on all tables in schema %s to steampipe_users;`, localSchema),
-
-		// Import the foreign schema into this connection.
-		fmt.Sprintf(`import foreign schema "%s" from server steampipe into %s;`, remoteSchema, localSchema),
-	}
-}
-
-func commentsQuery(p *steampipeconfig.ConnectionPlugin) []string {
-	var statements []string
-	for t, schema := range p.Schema.Schema {
-		table := db_common.PgEscapeName(t)
-		schemaName := db_common.PgEscapeName(p.ConnectionName)
-		if schema.Description != "" {
-			tableDescription := db_common.PgEscapeString(schema.Description)
-			statements = append(statements, fmt.Sprintf("COMMENT ON FOREIGN TABLE %s.%s is %s;", schemaName, table, tableDescription))
-		}
-		for _, c := range schema.Columns {
-			if c.Description != "" {
-				column := db_common.PgEscapeName(c.Name)
-				columnDescription := db_common.PgEscapeString(c.Description)
-				statements = append(statements, fmt.Sprintf("COMMENT ON COLUMN %s.%s.%s is %s;", schemaName, table, column, columnDescription))
-			}
-		}
-	}
-	return statements
-}
-
-func deleteConnectionQuery(name string) []string {
-	return []string{
-		fmt.Sprintf(`DROP SCHEMA IF EXISTS %s CASCADE;`, name),
-	}
-}
-
-func executeConnectionQueries(schemaQueries []string, updates *steampipeconfig.ConnectionUpdates) error {
-	log.Printf("[TRACE] there are connections to update\n")
-	_, err := executeSqlAsRoot(schemaQueries...)
-	if err != nil {
-		return err
-	}
-
-	// now update the state file
-	err = steampipeconfig.SaveConnectionState(updates.RequiredConnections)
-	if err != nil {
-		return err
-	}
-	return nil
-}

db/db_local/local_db_client_connections.go

@@ -0,0 +1,209 @@
+package db_local
+
+import (
+	"fmt"
+	"log"
+	"strings"
+
+	"github.com/turbot/steampipe/db/db_common"
+	"github.com/turbot/steampipe/steampipeconfig"
+	"github.com/turbot/steampipe/utils"
+)
+
+// RefreshConnections loads required connections from config
+// and update the database schema and search path to reflect the required connections
+// return whether any changes have been made
+func (c *LocalDbClient) refreshConnections() *steampipeconfig.RefreshConnectionResult {
+	res := &steampipeconfig.RefreshConnectionResult{}
+	utils.LogTime("db.refreshConnections start")
+	defer utils.LogTime("db.refreshConnections end")
+
+	// get a list of all existing schema names
+	schemaNames := c.client.SchemaMetadata().GetSchemas()
+
+	// determine any necessary connection updates
+	connectionUpdates, res := steampipeconfig.NewConnectionUpdates(schemaNames)
+	if res.Error != nil {
+		return res
+	}
+
+	log.Printf("[TRACE] refreshConnections, updates: %+v\n", connectionUpdates)
+
+	// if any plugins are missing, error for now but we could prompt for an install
+	missingCount := len(connectionUpdates.MissingPlugins)
+	if missingCount > 0 {
+		res.Error = fmt.Errorf("%d %s referenced in the connection config not installed: \n  %v",
+			missingCount,
+			utils.Pluralize("plugin", missingCount),
+			strings.Join(connectionUpdates.MissingPlugins, "\n  "))
+		return res
+	}
+
+	// now build list of necessary queries to perform the update
+	connectionQueries, otherRes := c.buildConnectionUpdateQueries(connectionUpdates)
+	// merge results into local results
+	res.Merge(otherRes)
+	if res.Error != nil {
+		return res
+	}
+
+	// if there are no connection queries, we are done
+	if len(connectionQueries) == 0 {
+		return res
+	}
+
+	// so there ARE connections to update
+
+	// execute the connection queries
+	if err := executeConnectionQueries(connectionQueries); err != nil {
+		res.Error = err
+		return res
+	}
+
+	// now serialise the connection state
+	// update required connections with the schema mode from the connection state and schema hash from the hash map
+	if err := steampipeconfig.SaveConnectionState(connectionUpdates.RequiredConnectionState); err != nil {
+		res.Error = err
+		return res
+	}
+
+	// reload the database schemas, since they have changed - otherwise we wouldn't be here
+	log.Println("[TRACE] RefreshConnections: reloading schema")
+	c.LoadSchema()
+
+	res.UpdatedConnections = true
+	return res
+
+}
+
+func (c *LocalDbClient) buildConnectionUpdateQueries(connectionUpdates *steampipeconfig.ConnectionUpdates) ([]string, *steampipeconfig.RefreshConnectionResult) {
+	var connectionQueries []string
+	var res *steampipeconfig.RefreshConnectionResult
+	numUpdates := len(connectionUpdates.Update)
+
+	log.Printf("[TRACE] buildConnectionUpdateQueries: num updates %d", numUpdates)
+
+	if numUpdates > 0 {
+
+		// find any plugins which use a newer sdk version than steampipe.
+		validationFailures, validatedUpdates, validatedPlugins := steampipeconfig.ValidatePlugins(connectionUpdates.Update, connectionUpdates.ConnectionPlugins)
+		if len(validationFailures) > 0 {
+			res.Warnings = append(res.Warnings, steampipeconfig.BuildValidationWarningString(validationFailures))
+		}
+
+		// get schema queries - this updates schemas for validated plugins and drops schemas for unvalidated plugins
+		connectionQueries = getSchemaQueries(validatedUpdates, validationFailures)
+		// add comments queries for validated connections
+		connectionQueries = append(connectionQueries, getCommentQueries(validatedPlugins)...)
+	}
+
+	for c := range connectionUpdates.Delete {
+		log.Printf("[TRACE] delete connection %s\n ", c)
+		connectionQueries = append(connectionQueries, deleteConnectionQuery(c)...)
+	}
+	return connectionQueries, res
+}
+
+func (c *LocalDbClient) updateConnectionMap() error {
+	// load the connection state and cache it!
+	log.Println("[TRACE]", "retrieving connection map")
+	connectionMap, err := steampipeconfig.GetConnectionState(c.client.SchemaMetadata().GetSchemas())
+	if err != nil {
+		return err
+	}
+	log.Println("[TRACE]", "setting connection map")
+	c.connectionMap = &connectionMap
+
+	return nil
+}
+
+func getSchemaQueries(updates steampipeconfig.ConnectionDataMap, failures []*steampipeconfig.ValidationFailure) []string {
+	var schemaQueries []string
+	for connectionName, plugin := range updates {
+		remoteSchema := steampipeconfig.PluginFQNToSchemaName(plugin.Plugin)
+		log.Printf("[TRACE] update connection %s, plugin Name %s, schema %s\n ", connectionName, plugin.Plugin, remoteSchema)
+		schemaQueries = append(schemaQueries, updateConnectionQuery(connectionName, remoteSchema)...)
+	}
+	for _, failure := range failures {
+		log.Printf("[TRACE] remove schema for conneciton failing validation connection %s, plugin Name %s\n ", failure.ConnectionName, failure.Plugin)
+		if failure.ShouldDropIfExists {
+			schemaQueries = append(schemaQueries, deleteConnectionQuery(failure.ConnectionName)...)
+		}
+	}
+
+	return schemaQueries
+}
+
+func getCommentQueries(plugins []*steampipeconfig.ConnectionPlugin) []string {
+	var commentQueries []string
+	for _, plugin := range plugins {
+		commentQueries = append(commentQueries, commentsQuery(plugin)...)
+	}
+	return commentQueries
+}
+
+func updateConnectionQuery(localSchema, remoteSchema string) []string {
+	// escape the name
+	localSchema = db_common.PgEscapeName(localSchema)
+	return []string{
+
+		// Each connection has a unique schema. The schema, and all objects inside it,
+		// are owned by the root user.
+		fmt.Sprintf(`drop schema if exists %s cascade;`, localSchema),
+		fmt.Sprintf(`create schema %s;`, localSchema),
+		fmt.Sprintf(`comment on schema %s is 'steampipe plugin: %s';`, localSchema, remoteSchema),
+
+		// Steampipe users are allowed to use the new schema
+		fmt.Sprintf(`grant usage on schema %s to steampipe_users;`, localSchema),
+
+		// Permissions are limited to select only, and should be granted for all new
+		// objects. Steampipe users cannot create tables or modify data in the
+		// connection schema - they need to use the public schema for that.  These
+		// commands alter the defaults for any objects created in the future.
+		// See https://www.postgresql.org/docs/12/ddl-priv.html
+		fmt.Sprintf(`alter default privileges in schema %s grant select on tables to steampipe_users;`, localSchema),
+
+		// If there are any objects already then grant their permissions now. (This
+		// should not actually do anything at this point.)
+		fmt.Sprintf(`grant select on all tables in schema %s to steampipe_users;`, localSchema),
+
+		// Import the foreign schema into this connection.
+		fmt.Sprintf(`import foreign schema "%s" from server steampipe into %s;`, remoteSchema, localSchema),
+	}
+}
+
+func commentsQuery(p *steampipeconfig.ConnectionPlugin) []string {
+	var statements []string
+	for t, schema := range p.Schema.Schema {
+		table := db_common.PgEscapeName(t)
+		schemaName := db_common.PgEscapeName(p.ConnectionName)
+		if schema.Description != "" {
+			tableDescription := db_common.PgEscapeString(schema.Description)
+			statements = append(statements, fmt.Sprintf("COMMENT ON FOREIGN TABLE %s.%s is %s;", schemaName, table, tableDescription))
+		}
+		for _, c := range schema.Columns {
+			if c.Description != "" {
+				column := db_common.PgEscapeName(c.Name)
+				columnDescription := db_common.PgEscapeString(c.Description)
+				statements = append(statements, fmt.Sprintf("COMMENT ON COLUMN %s.%s.%s is %s;", schemaName, table, column, columnDescription))
+			}
+		}
+	}
+	return statements
+}
+
+func deleteConnectionQuery(name string) []string {
+	return []string{
+		fmt.Sprintf(`DROP SCHEMA IF EXISTS %s CASCADE;`, name),
+	}
+}
+
+func executeConnectionQueries(schemaQueries []string) error {
+	log.Printf("[TRACE] there are connections to update\n")
+	_, err := executeSqlAsRoot(schemaQueries...)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

go.mod

@@ -26,6 +26,7 @@ require (
 	github.com/gin-gonic/gin v1.7.2
 	github.com/go-git/go-git/v5 v5.4.2
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.2
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.1
@@ -36,19 +37,16 @@ require (
 	github.com/hashicorp/terraform v0.15.1
 	github.com/jedib0t/go-pretty/v6 v6.0.6
 	github.com/karrick/gows v0.3.0
-	github.com/kr/text v0.2.0 // indirect
 	github.com/lib/pq v1.8.0
 	github.com/logrusorgru/aurora v2.0.3+incompatible
 	github.com/mattn/go-isatty v0.0.12
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db
-	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/olekukonko/tablewriter v0.0.4
 	github.com/opencontainers/image-spec v1.0.1
 	github.com/otiai10/copy v1.2.0
 	github.com/prometheus/client_golang v1.7.1 // indirect
 	github.com/prometheus/procfs v0.6.0 // indirect
-	github.com/sergi/go-diff v1.1.0 // indirect
 	github.com/shiena/ansicolor v0.0.0-20200904210342-c7312218db18
 	github.com/shirou/gopsutil v3.20.11+incompatible
 	github.com/sirupsen/logrus v1.8.1
@@ -67,9 +65,13 @@ require (
 	golang.org/x/text v0.3.7
 	google.golang.org/genproto v0.0.0-20211005153810-c76a74d43a8e // indirect
 	google.golang.org/grpc v1.41.0
+	google.golang.org/protobuf v1.27.1
 	gopkg.in/olahol/melody.v1 v1.0.0-20170518105555-d52139073376
 	rsc.io/letsencrypt v0.0.3 // indirect
 	sigs.k8s.io/yaml v1.1.0
 )
 
 replace github.com/c-bata/go-prompt => github.com/turbot/go-prompt v0.2.6-steampipe.0.20210830083819-c872df2bdcc9
+
+// schemamode
+replace github.com/turbot/steampipe-plugin-sdk => github.com/turbot/steampipe-plugin-sdk v1.6.3-0.20211008130330-3b06e80f00ee

plugin_manager/proto/plugin_manager.pb.go

@@ -0,0 +1,405 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.25.0
+// 	protoc        v3.13.0
+// source: plugin_manager.proto
+
+package proto
+
+import (
+	context "context"
+	proto "github.com/golang/protobuf/proto"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// This is a compile-time assertion that a sufficiently up-to-date version
+// of the legacy proto package is being used.
+const _ = proto.ProtoPackageIsVersion4
+
+type GetPluginRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Connection string `protobuf:"bytes,1,opt,name=connection,proto3" json:"connection,omitempty"`
+}
+
+func (x *GetPluginRequest) Reset() {
+	*x = GetPluginRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_plugin_manager_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *GetPluginRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetPluginRequest) ProtoMessage() {}
+
+func (x *GetPluginRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_plugin_manager_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use GetPluginRequest.ProtoReflect.Descriptor instead.
+func (*GetPluginRequest) Descriptor() ([]byte, []int) {
+	return file_plugin_manager_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *GetPluginRequest) GetConnection() string {
+	if x != nil {
+		return x.Connection
+	}
+	return ""
+}
+
+type GetPluginResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Protocol        string   `protobuf:"bytes,1,opt,name=Protocol,proto3" json:"Protocol,omitempty"`
+	ProtocolVersion int64    `protobuf:"varint,2,opt,name=ProtocolVersion,proto3" json:"ProtocolVersion,omitempty"`
+	Addr            *NetAddr `protobuf:"bytes,3,opt,name=Addr,proto3" json:"Addr,omitempty"`
+	Pid             int64    `protobuf:"varint,4,opt,name=Pid,proto3" json:"Pid,omitempty"`
+}
+
+func (x *GetPluginResponse) Reset() {
+	*x = GetPluginResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_plugin_manager_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *GetPluginResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetPluginResponse) ProtoMessage() {}
+
+func (x *GetPluginResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_plugin_manager_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use GetPluginResponse.ProtoReflect.Descriptor instead.
+func (*GetPluginResponse) Descriptor() ([]byte, []int) {
+	return file_plugin_manager_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *GetPluginResponse) GetProtocol() string {
+	if x != nil {
+		return x.Protocol
+	}
+	return ""
+}
+
+func (x *GetPluginResponse) GetProtocolVersion() int64 {
+	if x != nil {
+		return x.ProtocolVersion
+	}
+	return 0
+}
+
+func (x *GetPluginResponse) GetAddr() *NetAddr {
+	if x != nil {
+		return x.Addr
+	}
+	return nil
+}
+
+func (x *GetPluginResponse) GetPid() int64 {
+	if x != nil {
+		return x.Pid
+	}
+	return 0
+}
+
+type NetAddr struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Network string `protobuf:"bytes,1,opt,name=Network,proto3" json:"Network,omitempty"` // name of the network (for example, "tcp", "udp")
+	Address string `protobuf:"bytes,2,opt,name=Address,proto3" json:"Address,omitempty"` // string form of address (for example, "192.0.2.1:25", "[2001:db8::1]:80")
+}
+
+func (x *NetAddr) Reset() {
+	*x = NetAddr{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_plugin_manager_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *NetAddr) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*NetAddr) ProtoMessage() {}
+
+func (x *NetAddr) ProtoReflect() protoreflect.Message {
+	mi := &file_plugin_manager_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use NetAddr.ProtoReflect.Descriptor instead.
+func (*NetAddr) Descriptor() ([]byte, []int) {
+	return file_plugin_manager_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *NetAddr) GetNetwork() string {
+	if x != nil {
+		return x.Network
+	}
+	return ""
+}
+
+func (x *NetAddr) GetAddress() string {
+	if x != nil {
+		return x.Address
+	}
+	return ""
+}
+
+var File_plugin_manager_proto protoreflect.FileDescriptor
+
+var file_plugin_manager_proto_rawDesc = []byte{
+	0x0a, 0x14, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x5f, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x05, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x32, 0x0a,
+	0x10, 0x47, 0x65, 0x74, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x22, 0x8f, 0x01, 0x0a, 0x11, 0x47, 0x65, 0x74, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x52,
+	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x50, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x12, 0x28, 0x0a, 0x0f, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x56,
+	0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0f, 0x50, 0x72,
+	0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x22, 0x0a,
+	0x04, 0x41, 0x64, 0x64, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x2e, 0x4e, 0x65, 0x74, 0x41, 0x64, 0x64, 0x72, 0x52, 0x04, 0x41, 0x64, 0x64,
+	0x72, 0x12, 0x10, 0x0a, 0x03, 0x50, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x03, 0x52, 0x03,
+	0x50, 0x69, 0x64, 0x22, 0x3d, 0x0a, 0x07, 0x4e, 0x65, 0x74, 0x41, 0x64, 0x64, 0x72, 0x12, 0x18,
+	0x0a, 0x07, 0x4e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x07, 0x4e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x12, 0x18, 0x0a, 0x07, 0x41, 0x64, 0x64, 0x72,
+	0x65, 0x73, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x41, 0x64, 0x64, 0x72, 0x65,
+	0x73, 0x73, 0x32, 0x51, 0x0a, 0x0d, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x4d, 0x61, 0x6e, 0x61,
+	0x67, 0x65, 0x72, 0x12, 0x40, 0x0a, 0x09, 0x47, 0x65, 0x74, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e,
+	0x12, 0x17, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x50, 0x6c, 0x75, 0x67,
+	0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x2e, 0x47, 0x65, 0x74, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x09, 0x5a, 0x07, 0x2e, 0x3b, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_plugin_manager_proto_rawDescOnce sync.Once
+	file_plugin_manager_proto_rawDescData = file_plugin_manager_proto_rawDesc
+)
+
+func file_plugin_manager_proto_rawDescGZIP() []byte {
+	file_plugin_manager_proto_rawDescOnce.Do(func() {
+		file_plugin_manager_proto_rawDescData = protoimpl.X.CompressGZIP(file_plugin_manager_proto_rawDescData)
+	})
+	return file_plugin_manager_proto_rawDescData
+}
+
+var file_plugin_manager_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_plugin_manager_proto_goTypes = []interface{}{
+	(*GetPluginRequest)(nil),  // 0: proto.GetPluginRequest
+	(*GetPluginResponse)(nil), // 1: proto.GetPluginResponse
+	(*NetAddr)(nil),           // 2: proto.NetAddr
+}
+var file_plugin_manager_proto_depIdxs = []int32{
+	2, // 0: proto.GetPluginResponse.Addr:type_name -> proto.NetAddr
+	0, // 1: proto.PluginManager.GetPlugin:input_type -> proto.GetPluginRequest
+	1, // 2: proto.PluginManager.GetPlugin:output_type -> proto.GetPluginResponse
+	2, // [2:3] is the sub-list for method output_type
+	1, // [1:2] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_plugin_manager_proto_init() }
+func file_plugin_manager_proto_init() {
+	if File_plugin_manager_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_plugin_manager_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*GetPluginRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_plugin_manager_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*GetPluginResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_plugin_manager_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*NetAddr); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_plugin_manager_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   3,
+			NumExtensions: 0,
+			NumServices:   1,
+		},
+		GoTypes:           file_plugin_manager_proto_goTypes,
+		DependencyIndexes: file_plugin_manager_proto_depIdxs,
+		MessageInfos:      file_plugin_manager_proto_msgTypes,
+	}.Build()
+	File_plugin_manager_proto = out.File
+	file_plugin_manager_proto_rawDesc = nil
+	file_plugin_manager_proto_goTypes = nil
+	file_plugin_manager_proto_depIdxs = nil
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConnInterface
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion6
+
+// PluginManagerClient is the client API for PluginManager service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type PluginManagerClient interface {
+	GetPlugin(ctx context.Context, in *GetPluginRequest, opts ...grpc.CallOption) (*GetPluginResponse, error)
+}
+
+type pluginManagerClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewPluginManagerClient(cc grpc.ClientConnInterface) PluginManagerClient {
+	return &pluginManagerClient{cc}
+}
+
+func (c *pluginManagerClient) GetPlugin(ctx context.Context, in *GetPluginRequest, opts ...grpc.CallOption) (*GetPluginResponse, error) {
+	out := new(GetPluginResponse)
+	err := c.cc.Invoke(ctx, "/proto.PluginManager/GetPlugin", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// PluginManagerServer is the server API for PluginManager service.
+type PluginManagerServer interface {
+	GetPlugin(context.Context, *GetPluginRequest) (*GetPluginResponse, error)
+}
+
+// UnimplementedPluginManagerServer can be embedded to have forward compatible implementations.
+type UnimplementedPluginManagerServer struct {
+}
+
+func (*UnimplementedPluginManagerServer) GetPlugin(context.Context, *GetPluginRequest) (*GetPluginResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetPlugin not implemented")
+}
+
+func RegisterPluginManagerServer(s *grpc.Server, srv PluginManagerServer) {
+	s.RegisterService(&_PluginManager_serviceDesc, srv)
+}
+
+func _PluginManager_GetPlugin_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(GetPluginRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(PluginManagerServer).GetPlugin(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/proto.PluginManager/GetPlugin",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(PluginManagerServer).GetPlugin(ctx, req.(*GetPluginRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+var _PluginManager_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "proto.PluginManager",
+	HandlerType: (*PluginManagerServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "GetPlugin",
+			Handler:    _PluginManager_GetPlugin_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "plugin_manager.proto",
+}

schema/schema.go

@@ -14,7 +14,7 @@ func NewMetadata() *Metadata {
 	}
 }
 
-// Metadata :: struct to represent the schema of the database
+// Metadata is a struct to represent the schema of the database
 type Metadata struct {
 	// map {schemaname, {map {tablename -> tableschema}}
 	Schemas map[string]map[string]TableSchema
@@ -24,7 +24,7 @@ type Metadata struct {
 	TemporarySchemaName string
 }
 
-// TableSchema :: contains the details of a single table in the schema
+// TableSchema contains the details of a single table in the schema
 type TableSchema struct {
 	// map {columnName -> columnschema}
 	Columns     map[string]ColumnSchema
@@ -33,7 +33,7 @@ type TableSchema struct {
 	Description string
 }
 
-// ColumnSchema :: contains the details of a single column in a table
+// ColumnSchema contains the details of a single column in a table
 type ColumnSchema struct {
 	ID          string
 	Name        string
@@ -43,7 +43,7 @@ type ColumnSchema struct {
 	Description string
 }
 
-// GetSchemas :: returns all foreign schema names
+// GetSchemas returns all foreign schema names
 func (m *Metadata) GetSchemas() []string {
 	schemas := []string{}
 	for schema := range m.Schemas {

sp

@@ -0,0 +1 @@
+/usr/local/bin/sp

steampipeconfig/connection_data.go

@@ -0,0 +1,33 @@
+package steampipeconfig
+
+import (
+	"github.com/turbot/steampipe/steampipeconfig/modconfig"
+)
+
+// ConnectionData is a struct containing all details for a connection
+// - the plugin name and checksum, the connection config and options
+// json tags needed as this is stored in the connection state file
+type ConnectionData struct {
+	// the fully qualified name of the plugin
+	Plugin string
+	// the checksum of the plugin file
+	CheckSum string
+	// the underlying connection object
+	Connection *modconfig.Connection
+	// schema mode - static or dynamic
+	SchemaMode string `json:"SchemaMode,omitempty"`
+	// the hash of the connection schema
+	SchemaHash string `json:"SchemaHash,omitempty"`
+}
+
+func (p *ConnectionData) Equals(other *ConnectionData) bool {
+	if p.Connection == nil || other.Connection == nil {
+		// if either object has a nil Connection, then it may be data from an old connection state file
+		// return false, so that connections get refreshed and this file gets written in the new format in the process
+		return false
+	}
+
+	return p.Plugin == other.Plugin &&
+		p.CheckSum == other.CheckSum &&
+		p.Connection.Equals(other.Connection)
+}

steampipeconfig/connection_data_map.go

@@ -0,0 +1,70 @@
+package steampipeconfig
+
+import (
+	"fmt"
+
+	"github.com/turbot/go-kit/helpers"
+	"github.com/turbot/steampipe/steampipeconfig/modconfig"
+	"github.com/turbot/steampipe/utils"
+)
+
+type ConnectionDataMap map[string]*ConnectionData
+
+func (m ConnectionDataMap) Equals(other ConnectionDataMap) bool {
+	if m != nil && other == nil {
+		return false
+	}
+	for k, lVal := range m {
+		rVal, ok := other[k]
+		if !ok || !lVal.Equals(rVal) {
+			return false
+		}
+	}
+	for k := range other {
+		if _, ok := m[k]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+// NewConnectionDataMap tries to populate a map of connection data for all connections in connectionMap
+func NewConnectionDataMap(connectionMap map[string]*modconfig.Connection) (ConnectionDataMap, []string, error) {
+	utils.LogTime("steampipeconfig.getRequiredConnections start")
+	defer utils.LogTime("steampipeconfig.getRequiredConnections end")
+
+	requiredConnections := ConnectionDataMap{}
+	var missingPlugins []string
+
+	utils.LogTime("steampipeconfig.getRequiredConnections config-iteration start")
+	// populate checksum for each referenced plugin
+	for name, connection := range connectionMap {
+		remoteSchema := connection.Plugin
+		pluginPath, err := GetPluginPath(connection)
+		if err != nil {
+			err := fmt.Errorf("failed to load connection '%s': %v\n%s", connection.Name, err, connection.DeclRange)
+			return nil, nil, err
+		}
+		// if plugin is not installed, the path will be returned as empty
+		if pluginPath == "" {
+			if !helpers.StringSliceContains(missingPlugins, connection.Plugin) {
+				missingPlugins = append(missingPlugins, connection.Plugin)
+			}
+			continue
+		}
+
+		checksum, err := utils.FileHash(pluginPath)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		requiredConnections[name] = &ConnectionData{
+			Plugin:     remoteSchema,
+			CheckSum:   checksum,
+			Connection: connection,
+		}
+	}
+	utils.LogTime("steampipeconfig.getRequiredConnections config-iteration end")
+
+	return requiredConnections, missingPlugins, nil
+}

steampipeconfig/connection_state.go

@@ -10,8 +10,8 @@ import (
 	"github.com/turbot/steampipe/utils"
 )
 
-// GetConnectionState :: load connection state file, and remove any connections which do not exist in the db
-func GetConnectionState(schemas []string) (ConnectionDataMap, error) {
+// GetConnectionState loads the connection state file, and remove any connections which do not exist in the db
+func GetConnectionState(schemaNames []string) (ConnectionDataMap, error) {
 	utils.LogTime("steampipeconfig.GetConnectionState start")
 	defer utils.LogTime("steampipeconfig.GetConnectionState end")
 
@@ -20,7 +20,7 @@ func GetConnectionState(schemas []string) (ConnectionDataMap, error) {
 	if err != nil {
 		return nil, err
 	}
-	return pruneConnectionState(connectionState, schemas), nil
+	return pruneConnectionState(connectionState, schemaNames), nil
 }
 
 // load and parse the connection config
@@ -45,9 +45,9 @@ func loadConnectionStateFile() (ConnectionDataMap, error) {
 }
 
 // update connection map to remove any connections which do not exist in the list of given schemas
-func pruneConnectionState(connections ConnectionDataMap, schemas []string) ConnectionDataMap {
+func pruneConnectionState(connections ConnectionDataMap, schemaNames []string) ConnectionDataMap {
 	var actualConnectionState = make(ConnectionDataMap)
-	for _, connectionName := range schemas {
+	for _, connectionName := range schemaNames {
 		if connection, ok := connections[connectionName]; ok {
 			actualConnectionState[connectionName] = connection
 		}

steampipeconfig/connection_test.go

@@ -39,7 +39,7 @@ var testCasesGetConnectionsToUpdate = map[string]getConnectionsToUpdateTest{
 				Connection: &modconfig.Connection{Name: "a"},
 			},
 		},
-		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnections: ConnectionDataMap{
+		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnectionState: ConnectionDataMap{
 			"a": {
 				Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 				CheckSum:   connectionTest1Checksum,
@@ -69,7 +69,7 @@ connection "b" {
 				Connection: &modconfig.Connection{Name: "b"},
 			},
 		},
-		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnections: ConnectionDataMap{
+		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnectionState: ConnectionDataMap{
 			"a": {
 				Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 				CheckSum:   connectionTest1Checksum,
@@ -104,7 +104,7 @@ connection "b" {
 				Connection: &modconfig.Connection{Name: "b"},
 			},
 		},
-		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnections: ConnectionDataMap{
+		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnectionState: ConnectionDataMap{
 			"a": {
 				Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 				CheckSum:   connectionTest1Checksum,
@@ -138,7 +138,7 @@ connection "b" {
 				Connection: &modconfig.Connection{Name: "b"},
 			},
 		},
-		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnections: ConnectionDataMap{
+		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnectionState: ConnectionDataMap{
 			"a": {
 				Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 				CheckSum:   connectionTest1Checksum,
@@ -172,7 +172,7 @@ connection "b" {
 				Connection: &modconfig.Connection{Name: "b"},
 			},
 		},
-		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnections: ConnectionDataMap{
+		expected: &ConnectionUpdates{Update: ConnectionDataMap{}, Delete: ConnectionDataMap{}, RequiredConnectionState: ConnectionDataMap{
 			"a": {
 				Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 				CheckSum:   connectionTest1Checksum,
@@ -204,7 +204,7 @@ connection "b" {
 				CheckSum:   connectionTest1Checksum,
 				Connection: &modconfig.Connection{Name: "a"},
 			},
-		}, Delete: ConnectionDataMap{}, RequiredConnections: ConnectionDataMap{
+		}, Delete: ConnectionDataMap{}, RequiredConnectionState: ConnectionDataMap{
 			"a": {
 				Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 				CheckSum:   connectionTest1Checksum,
@@ -248,7 +248,7 @@ connection "b" {
 			},
 		},
 			Delete: ConnectionDataMap{},
-			RequiredConnections: ConnectionDataMap{
+			RequiredConnectionState: ConnectionDataMap{
 				"a": {
 					Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 					CheckSum:   connectionTest1Checksum,
@@ -295,7 +295,7 @@ connection "b" {
 				Connection: &modconfig.Connection{Name: "b"},
 			},
 		}, Delete: ConnectionDataMap{},
-			RequiredConnections: ConnectionDataMap{
+			RequiredConnectionState: ConnectionDataMap{
 				"a": {
 					Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 					CheckSum:   connectionTest1Checksum,
@@ -342,7 +342,7 @@ connection "b" {
 			},
 		},
 			Delete: ConnectionDataMap{},
-			RequiredConnections: ConnectionDataMap{
+			RequiredConnectionState: ConnectionDataMap{
 				"a": {
 					Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 					CheckSum:   connectionTest1Checksum,
@@ -390,7 +390,7 @@ connection "b" {
 				},
 			},
 			Delete: ConnectionDataMap{},
-			RequiredConnections: ConnectionDataMap{
+			RequiredConnectionState: ConnectionDataMap{
 				"a": {
 					Plugin:     "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
 					CheckSum:   connectionTest1Checksum,
@@ -429,19 +429,19 @@ func TestGetConnectionsToUpdate(t *testing.T) {
 		if config == nil {
 			t.Fatalf("Could not load config")
 		}
-		requiredConnections := config.Connections
+		GlobalConfig = config
 		// all tests assume connections a, b
-		res, err := GetConnectionsToUpdate([]string{"a", "b"}, requiredConnections)
+		updates, res := NewConnectionUpdates([]string{"a", "b"})
 
-		if err != nil && test.expected != "ERROR" {
+		if res.Error != nil && test.expected != "ERROR" {
 			continue
-			t.Fatalf("GetConnectionsToUpdate failed with unexpected error: %v", err)
+			t.Fatalf("NewConnectionUpdates failed with unexpected error: %v", err)
 		}
 
 		expectedUpdates := test.expected.(*ConnectionUpdates)
-		if !res.RequiredConnections.Equals(expectedUpdates.RequiredConnections) ||
-			!res.Update.Equals(expectedUpdates.Update) ||
-			!res.Delete.Equals(expectedUpdates.Delete) {
+		if !updates.RequiredConnectionState.Equals(expectedUpdates.RequiredConnectionState) ||
+			!updates.Update.Equals(expectedUpdates.Update) ||
+			!updates.Delete.Equals(expectedUpdates.Delete) {
 			t.Errorf(`Test: '%s'' FAILED`, name)
 
 		}

steampipeconfig/connection_updates.go

@@ -3,9 +3,12 @@ package steampipeconfig
 import (
 	"fmt"
 	"log"
+	"sort"
+	"strings"
+	"time"
 
-	"github.com/turbot/go-kit/helpers"
-	"github.com/turbot/steampipe/steampipeconfig/modconfig"
+	"github.com/turbot/steampipe-plugin-sdk/grpc/proto"
+	"github.com/turbot/steampipe-plugin-sdk/plugin"
 	"github.com/turbot/steampipe/utils"
 )
 
@@ -14,142 +17,222 @@ type ConnectionUpdates struct {
 	Delete         ConnectionDataMap
 	MissingPlugins []string
 	// the connections which will exist after the update
-	RequiredConnections ConnectionDataMap
+	RequiredConnectionState ConnectionDataMap
+	// connection plugins required to perform the updates
+	ConnectionPlugins      map[string]*ConnectionPlugin
+	currentConnectionState ConnectionDataMap
 }
 
-func newConnectionUpdates() *ConnectionUpdates {
-	return &ConnectionUpdates{
-		Update:              ConnectionDataMap{},
-		Delete:              ConnectionDataMap{},
-		MissingPlugins:      []string{},
-		RequiredConnections: ConnectionDataMap{},
-	}
-}
+// NewConnectionUpdates returns updates to be made to the database to sync with connection config
+func NewConnectionUpdates(schemaNames []string) (*ConnectionUpdates, *RefreshConnectionResult) {
+	utils.LogTime("NewConnectionUpdates start")
+	defer utils.LogTime("NewConnectionUpdates end")
 
-// ConnectionData is a struct containing all details for a connection - the plugin name and checksum, the connection config and options
-type ConnectionData struct {
-	// the fully qualified name of the plugin
-	Plugin string
-	// the checksum of the plugin file
-	CheckSum string
-	// the underlying connection object
-	Connection *modconfig.Connection
-}
-
-func (p *ConnectionData) Equals(other *ConnectionData) bool {
-	if p.Connection == nil || other.Connection == nil {
-		// if any one them has a `nil` Connection, then
-		// this is data from an old connection state file.
-		// return false, so that connections get refreshed
-		// and this file gets written in the new format in the process
-		return false
+	res := &RefreshConnectionResult{}
+	// build connection data for all required connections
+	requiredConnectionState, missingPlugins, err := NewConnectionDataMap(GlobalConfig.Connections)
+	if err != nil {
+		res.Error = err
+		return nil, res
 	}
 
-	return p.Plugin == other.Plugin &&
-		p.CheckSum == other.CheckSum &&
-		p.Connection.Equals(other.Connection)
-}
-
-type ConnectionDataMap map[string]*ConnectionData
-
-func (m ConnectionDataMap) Equals(other ConnectionDataMap) bool {
-	if m != nil && other == nil {
-		return false
+	updates := &ConnectionUpdates{
+		Update:                  ConnectionDataMap{},
+		Delete:                  ConnectionDataMap{},
+		MissingPlugins:          missingPlugins,
+		RequiredConnectionState: requiredConnectionState,
 	}
-	for k, lVal := range m {
-		rVal, ok := other[k]
-		if !ok {
-			return false
-		}
-		if !lVal.Equals(rVal) {
-			return false
-		}
-	}
-	for k := range other {
-		if _, ok := m[k]; !ok {
-			return false
-		}
-	}
-	return true
-}
-
-// GetConnectionsToUpdate :: returns updates to be made to the database to sync with connection config
-func GetConnectionsToUpdate(schemas []string, connectionConfig map[string]*modconfig.Connection) (*ConnectionUpdates, error) {
-	utils.LogTime("steampipeconfig.GetConnectionsToUpdate start")
-	defer utils.LogTime("steampipeconfig.GetConnectionsToUpdate end")
 
 	// load the connection state file and filter out any connections which are not in the list of schemas
 	// this allows for the database being rebuilt,modified externally
-	connectionState, err := GetConnectionState(schemas)
+	connectionState, err := GetConnectionState(schemaNames)
 	if err != nil {
-		return nil, err
+		res.Error = err
+		return nil, res
 	}
+	updates.currentConnectionState = connectionState
 
-	requiredConnections, missingPlugins, err := getRequiredConnections(connectionConfig)
-	if err != nil {
-		return nil, err
+	// for any connections with dynamic schema, we need to reload their schema
+	// instantiate connection plugins for all connections with dynamic schema - this will retrieve their current schema
+	dynamicSchemaHashMap, connectionsPluginsWithDynamicSchema, res := getSchemaHashesForDynamicSchemas(requiredConnectionState, connectionState)
+	if res.Error != nil {
+		return nil, res
 	}
 
-	result := newConnectionUpdates()
-	result.MissingPlugins = missingPlugins
-	result.RequiredConnections = requiredConnections
-
 	// connections to create/update
-	for connection, requiredPlugin := range requiredConnections {
-		current, ok := connectionState[connection]
-		if !ok || !current.Equals(requiredPlugin) {
-			log.Printf("[TRACE] connection %s is out of date or missing\n", connection)
-			result.Update[connection] = requiredPlugin
+	for name, requiredConnectionData := range requiredConnectionState {
+		// check whether this connection exists in the state
+		currentConnectionData, ok := connectionState[name]
+		// if it does not exist, or is not equal, add to updates
+		if !ok || !currentConnectionData.Equals(requiredConnectionData) {
+			log.Printf("[TRACE] connection %s is out of date or missing\n", name)
+			updates.Update[name] = requiredConnectionData
 		}
 	}
 
-	// connections to delete
+	// connections to delete - any connection which is in connection state but NOT required connections
 	for connection, requiredPlugin := range connectionState {
-		if _, ok := requiredConnections[connection]; !ok {
+		if _, ok := requiredConnectionState[connection]; !ok {
 			log.Printf("[TRACE] connection %s is no longer required\n", connection)
-			result.Delete[connection] = requiredPlugin
+			updates.Delete[connection] = requiredPlugin
 		}
 	}
-	return result, nil
+
+	// now for every connection with dynamic schema, check whether the schema we have just fetched
+	// matches the existing db schema
+	for name, requiredHash := range dynamicSchemaHashMap {
+		// get the connection data from the loaded connection state
+		connectionData, ok := connectionState[name]
+		// if the connection exists in the state, does the schemas hash match?
+		if ok && connectionData.SchemaHash != requiredHash {
+			updates.Update[name] = connectionData
+		}
+	}
+
+	//  instantiate connection plugins for all updates
+	// NOTE - we may have already created some connection plugins (if they have dynamic schema)
+	// - pass in the list of connection plugins we have already loaded
+
+	connectionPlugins, otherRes := createConnectionPlugins(updates.Update, connectionsPluginsWithDynamicSchema)
+	// merge results into local results
+	res.Merge(otherRes)
+	if res.Error != nil {
+		return nil, res
+	}
+
+	updates.ConnectionPlugins = connectionPlugins
+	// set the schema mode and hash on the connection data in required state
+	updates.updateRequiredStateWithSchemaProperties(dynamicSchemaHashMap)
+
+	return updates, res
 }
 
-// load and parse the connection config
-func getRequiredConnections(connectionConfig map[string]*modconfig.Connection) (ConnectionDataMap, []string, error) {
-	utils.LogTime("steampipeconfig.getRequiredConnections start")
-	defer utils.LogTime("steampipeconfig.getRequiredConnections end")
-
-	requiredConnections := ConnectionDataMap{}
-	var missingPlugins []string
-
-	utils.LogTime("steampipeconfig.getRequiredConnections config-iteration start")
-	// populate checksum for each referenced plugin
-	for name, connection := range connectionConfig {
-		remoteSchema := connection.Plugin
-		pluginPath, err := GetPluginPath(connection)
-		if err != nil {
-			err := fmt.Errorf("failed to load connection '%s': %v\n%s", connection.Name, err, connection.DeclRange)
-			return nil, nil, err
+// update requiredConnections - set the schema hash and schema mode for all elements of RequiredConnectionState
+// default to the existing state, but if anm update is required, get the updated value
+func (u *ConnectionUpdates) updateRequiredStateWithSchemaProperties(schemaHashMap map[string]string) {
+	// we only need to update connections which are being updated
+	for k, v := range u.RequiredConnectionState {
+		if currentConectionState, ok := u.currentConnectionState[k]; ok {
+			v.SchemaHash = currentConectionState.SchemaHash
+			v.SchemaMode = currentConectionState.SchemaMode
 		}
-		// if plugin is not installed, the path will be returned as empty
-		if pluginPath == "" {
-			if !helpers.StringSliceContains(missingPlugins, connection.Plugin) {
-				missingPlugins = append(missingPlugins, connection.Plugin)
+		// if the schemaHashMap contains this connection, use that value
+		if schemaHash, ok := schemaHashMap[k]; ok {
+			v.SchemaHash = schemaHash
+		}
+		// have we loaded a connection plugin for this connection
+		// - if so us the schema mode from the schema  it has loaded
+		if connectionPlugin, ok := u.ConnectionPlugins[k]; ok {
+			v.SchemaMode = connectionPlugin.Schema.Mode
+			// if the schema hash is still not set, calculate the value from the conneciton plugin schema
+			// this will happen the first time we load a plugin
+			if v.SchemaHash == "" {
+				v.SchemaHash = pluginSchemaHash(connectionPlugin.Schema)
 			}
-			continue
 		}
 
-		checksum, err := utils.FileHash(pluginPath)
-		if err != nil {
-			return nil, nil, err
-		}
+	}
+}
 
-		requiredConnections[name] = &ConnectionData{
-			Plugin:     remoteSchema,
-			CheckSum:   checksum,
-			Connection: connection,
+func createConnectionPlugins(requiredConnections ConnectionDataMap, alreadyLoaded map[string]*ConnectionPlugin) (map[string]*ConnectionPlugin, *RefreshConnectionResult) {
+	if alreadyLoaded == nil {
+		alreadyLoaded = make(map[string]*ConnectionPlugin)
+	}
+	var connectionPlugins = make(map[string]*ConnectionPlugin)
+	res := &RefreshConnectionResult{}
+
+	// create channels buffered to hold all updates
+	numConnectionPlugins := len(requiredConnections)
+	var pluginChan = make(chan *ConnectionPlugin, numConnectionPlugins)
+	var errorChan = make(chan error, numConnectionPlugins)
+
+	// keep track of how many plugins we are creating
+	pluginCount := 0
+	for connectionName, connectionData := range requiredConnections {
+		// if we have already loaded this plugin, just send down the channel
+		if existingConnectionPlugin, ok := alreadyLoaded[connectionName]; ok {
+			pluginChan <- existingConnectionPlugin
+		} else {
+			pluginCount++
+			// instantiate the connection plugin, and retrieve schema
+			go getConnectionPluginAsync(connectionData, pluginChan, errorChan)
 		}
 	}
-	utils.LogTime("steampipeconfig.getRequiredConnections config-iteration end")
 
-	return requiredConnections, missingPlugins, nil
+	for i := 0; i < numConnectionPlugins; i++ {
+		select {
+		case err := <-errorChan:
+			log.Println("[TRACE] get connections err chan select - adding warning", "error", err)
+			res.Warnings = append(res.Warnings, err.Error())
+		case p := <-pluginChan:
+			connectionPlugins[p.ConnectionName] = p
+		case <-time.After(10 * time.Second):
+			res.Error = fmt.Errorf("timed out retrieving schema from plugins")
+			return nil, res
+		}
+	}
+
+	return connectionPlugins, res
+}
+
+func getConnectionPluginAsync(connectionData *ConnectionData, pluginChan chan *ConnectionPlugin, errorChan chan error) {
+	p, err := CreateConnectionPlugin(connectionData.Connection, true)
+	if err != nil {
+		errorChan <- err
+		return
+	}
+	pluginChan <- p
+
+	p.Plugin.Client.Kill()
+}
+
+func getSchemaHashesForDynamicSchemas(requiredConnectionData ConnectionDataMap, connectionState ConnectionDataMap) (map[string]string, map[string]*ConnectionPlugin, *RefreshConnectionResult) {
+	// for every required connection, check the connection state to determine whether it has a dynamic schema
+	// if we have never loaded the conneciton, there wil be no stste so we cannot retrieve this informaiton
+	// however in this case we will load the connection anyway
+
+	var connectionsWithDynamicSchema = make(ConnectionDataMap)
+	for requiredConnectionName, requiredConnection := range requiredConnectionData {
+		if existingConnection, ok := connectionState[requiredConnectionName]; ok {
+			// SchemaMode will be unpopulated for plugins using an older version of the sdk
+			// that is fine, we treat that as SchemaModeDynamic
+			if existingConnection.SchemaMode == plugin.SchemaModeDynamic {
+				connectionsWithDynamicSchema[requiredConnectionName] = requiredConnection
+			}
+		}
+	}
+
+	connectionsPluginsWithDynamicSchema, res := createConnectionPlugins(connectionsWithDynamicSchema, nil)
+	hashMap := make(map[string]string)
+	for name, c := range connectionsPluginsWithDynamicSchema {
+		// update schema hash stored in required connections so it is persisted in the state ius updates are made
+		schemaHash := pluginSchemaHash(c.Schema)
+		hashMap[name] = schemaHash
+	}
+	return hashMap, connectionsPluginsWithDynamicSchema, res
+}
+
+func pluginSchemaHash(s *proto.Schema) string {
+	var sb strings.Builder
+
+	// build ordered list of tables
+	var tables = make([]string, len(s.Schema))
+	idx := 0
+	for tableName := range s.Schema {
+		tables[idx] = tableName
+		idx++
+	}
+	sort.Strings(tables)
+
+	// now build  a string from the ordered table schemas
+	for _, tableName := range tables {
+		sb.WriteString(tableName)
+		tableSchema := s.Schema[tableName]
+		for _, c := range tableSchema.Columns {
+			sb.WriteString(c.Name)
+			sb.WriteString(fmt.Sprintf("%d", c.Type))
+		}
+	}
+	str := sb.String()
+	return utils.GetMD5Hash(str)
 }

steampipeconfig/connection_validation.go

@@ -26,7 +26,7 @@ func (v ValidationFailure) String() string {
 	)
 }
 
-func ValidatePlugins(updates ConnectionDataMap, plugins []*ConnectionPlugin) ([]*ValidationFailure, ConnectionDataMap, []*ConnectionPlugin) {
+func ValidatePlugins(updates ConnectionDataMap, plugins map[string]*ConnectionPlugin) ([]*ValidationFailure, ConnectionDataMap, []*ConnectionPlugin) {
 	var validatedPlugins []*ConnectionPlugin
 	var validatedUpdates = ConnectionDataMap{}
 

steampipeconfig/load_config.go

@@ -18,7 +18,7 @@ import (
 	"github.com/turbot/steampipe/utils"
 )
 
-var Config *SteampipeConfig
+var GlobalConfig *SteampipeConfig
 var defaultConfigFileName = "default.spc"
 
 // LoadSteampipeConfig loads the HCL connection config and workspace options

steampipeconfig/modconfig/connection.go

@@ -20,6 +20,7 @@ const (
 //
 // (Partial as the connection config, which is plugin specific, is stored as raw HCL.
 // This will be parsed by the plugin)
+// json tags needed as this is stored in the connection state file
 type Connection struct {
 	// connection name
 	Name string
@@ -28,18 +29,18 @@ type Connection struct {
 	// The fully qualified name of the plugin. derived from the short name
 	Plugin string
 	// Type - supported values: "aggregator"
-	Type string
+	Type string `json:"Type,omitempty"`
 	// this is a list of names or wildcards which are resolved to connections
 	// (only valid for "aggregator" type)
-	ConnectionNames []string
+	ConnectionNames []string `json:"Connections,omitempty"`
 	// a list of the resolved child connections
 	// (only valid for "aggregator" type)
-	Connections map[string]*Connection
+	Connections map[string]*Connection `json:"-"`
 	// unparsed HCL of plugin specific connection config
-	Config string
+	Config string `json:"Config,omitempty"`
 
 	// options
-	Options   *options.Connection
+	Options   *options.Connection `json:"Options,omitempty"`
 	DeclRange hcl.Range
 }
 
@@ -58,7 +59,7 @@ func (c *Connection) Equals(other *Connection) bool {
 	}
 	return c.Name == other.Name &&
 		connectionOptionsEqual &&
-		reflect.DeepEqual(c.Config, other.Config)
+		c.Config == other.Config
 }
 
 // SetOptions sets the options on the connection

steampipeconfig/options/connection.go

@@ -5,10 +5,11 @@ import (
 	"strings"
 )
 
-// options.Connection
+// Connection is a struct representing connection options
+// json tags needed as this is stored in the connection state file
 type Connection struct {
-	Cache    *bool `hcl:"cache"`
-	CacheTTL *int  `hcl:"cache_ttl"`
+	Cache    *bool `hcl:"cache" json:"Cache,omitempty"`
+	CacheTTL *int  `hcl:"cache_ttl" json:"CacheTTL,omitempty"`
 }
 
 func (c *Connection) ConfigMap() map[string]interface{} {
@@ -16,7 +17,7 @@ func (c *Connection) ConfigMap() map[string]interface{} {
 	return map[string]interface{}{}
 }
 
-// merge other options over the the top of this options object
+// Merge merges other options over the the top of this options object
 // i.e. if a property is set in otherOptions, it takes precedence
 func (c *Connection) Merge(otherOptions Options) {
 	switch o := otherOptions.(type) {
@@ -31,8 +32,7 @@ func (c *Connection) Merge(otherOptions Options) {
 }
 
 func (c *Connection) Equals(other *Connection) bool {
-	return c.Cache == other.Cache &&
-		*c.CacheTTL == *other.CacheTTL
+	return c.String() == other.String()
 }
 
 func (c *Connection) String() string {

steampipeconfig/refresh_connections_result.go

@@ -1,4 +1,4 @@
-package db_common
+package steampipeconfig
 
 import "fmt"
 
@@ -14,3 +14,16 @@ func (r *RefreshConnectionResult) ShowWarnings() {
 		fmt.Println(w)
 	}
 }
+
+func (r *RefreshConnectionResult) Merge(other *RefreshConnectionResult) {
+	if other == nil {
+		return
+	}
+	if other.UpdatedConnections {
+		r.UpdatedConnections = other.UpdatedConnections
+	}
+	if other.Error != nil {
+		r.Error = other.Error
+	}
+	r.Warnings = append(other.Warnings)
+}

steampipeconfig/steampipeconfig.go

@@ -53,7 +53,7 @@ func (c *SteampipeConfig) Validate() error {
 	return nil
 }
 
-// ConfigMap :: create a config map to pass to viper
+// ConfigMap creates a config map to pass to viper
 func (c *SteampipeConfig) ConfigMap() map[string]interface{} {
 	res := map[string]interface{}{}
 

steampipeconfig/test_data/connections_to_update/config/default.spc

@@ -0,0 +1,31 @@
+
+#
+# For detailed descriptions, see the reference documentation
+# at https://steampipe.io/docs/reference/cli-args
+#
+
+# options "connection" {
+#   cache     = true # true, false
+#   cache_ttl = 300  # expiration (TTL) in seconds
+# }
+
+# options "database" {
+#   port        = 9193    # any valid, open port number
+#   listen      = "local" # local, network
+#   search_path =  ""     # comma-separated string
+# }
+
+# options "terminal" {
+#   multi               = false   # true, false
+#   output              = "table" # json, csv, table, line
+#   header              = true    # true, false
+#   separator           = ","     # any single char
+#   timing              = false   # true, false
+#   search_path         =  ""     # comma-separated string
+#   search_path_prefix  =  ""     # comma-separated string
+#   watch  			    =  true   # true, false
+# }
+
+# options "general" {
+#   update_check = true # true, false
+# }

steampipeconfig/test_data/connections_to_update/config/test0.spc

@@ -0,0 +1,8 @@
+connection "a" {
+	 plugin = "connection-test-1"
+	}
+	
+	connection "b" {
+	 plugin = "connection-test-1"
+	}
+	

steampipeconfig/test_data/connections_to_update/internal/connection.json

@@ -0,0 +1,46 @@
+{
+ "a": {
+  "Plugin": "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
+  "CheckSum": "xxxxxx",
+  "Connection": {
+   "Name": "a",
+   "PluginShortName": "",
+   "Plugin": "",
+   "DeclRange": {
+    "Filename": "",
+    "Start": {
+     "Line": 0,
+     "Column": 0,
+     "Byte": 0
+    },
+    "End": {
+     "Line": 0,
+     "Column": 0,
+     "Byte": 0
+    }
+   }
+  }
+ },
+ "b": {
+  "Plugin": "hub.steampipe.io/plugins/turbot/connection-test-1@latest",
+  "CheckSum": "xxxxxx",
+  "Connection": {
+   "Name": "a",
+   "PluginShortName": "",
+   "Plugin": "",
+   "DeclRange": {
+    "Filename": "",
+    "Start": {
+     "Line": 0,
+     "Column": 0,
+     "Byte": 0
+    },
+    "End": {
+     "Line": 0,
+     "Column": 0,
+     "Byte": 0
+    }
+   }
+  }
+ }
+}

tests/acceptance/test_data/introspection_table_mod/control.sp

@@ -1,5 +0,0 @@
-control "sample_control_1"{
-    title ="Sample control 1"
-    description = "This is a sample control to test introspection tables features"
-    query = query.sample_query_1
-}

workspace/connection_watcher.go

@@ -66,7 +66,7 @@ func (w *ConnectionWatcher) handleFileWatcherEvent([]fsnotify.Event) {
 		utils.ShowError(err)
 		return
 	}
-	steampipeconfig.Config = config
+	steampipeconfig.GlobalConfig = config
 	refreshResult := w.client.RefreshConnectionAndSearchPaths()
 	if refreshResult.Error != nil {
 		fmt.Println()