Release Steampipe v0.23.5

CHANGELOG.md

@@ -1,3 +1,7 @@
+## v0.23.5 [2024-08-21]
+_Bug fixes_
+- Fix issue where refresh connections was not creating a new connection if it was not in the search path. ([#4353](https://github.com/turbot/steampipe/issues/4353))
+
 ## v0.23.4 [2024-08-13]
 _Whats new_
 - Compiled with Go 1.22. ([#4340](https://github.com/turbot/steampipe/issues/4340))

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/hashicorp/go-version v1.6.0
 	github.com/hashicorp/hcl/v2 v2.20.1
 	github.com/jackc/pgconn v1.14.3
-	github.com/jackc/pgx/v5 v5.5.5
+	github.com/jackc/pgx/v5 v5.6.0
 	github.com/jedib0t/go-pretty/v6 v6.5.9
 	github.com/karrick/gows v0.3.0
 	github.com/logrusorgru/aurora v2.0.3+incompatible
@@ -49,16 +49,16 @@ require (
 	github.com/stevenle/topsort v0.2.0
 	github.com/thediveo/enumflag/v2 v2.0.5
 	github.com/turbot/go-kit v0.10.0-rc.0
-	github.com/turbot/pipe-fittings v1.2.0
+	github.com/turbot/pipe-fittings v1.5.2
 	github.com/turbot/steampipe-cloud-sdk-go v0.6.0
 	github.com/turbot/steampipe-plugin-sdk/v5 v5.10.3
 	github.com/turbot/terraform-components v0.0.0-20231213122222-1f3526cab7a7
 	github.com/xlab/treeprint v1.2.0
 	github.com/zclconf/go-cty v1.14.4
 	github.com/zclconf/go-cty-yaml v1.0.3
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
-	golang.org/x/sync v0.7.0
-	golang.org/x/text v0.16.0
+	golang.org/x/sync v0.8.0
+	golang.org/x/text v0.17.0
 	google.golang.org/grpc v1.64.1
 	google.golang.org/protobuf v1.34.1
 	gopkg.in/olahol/melody.v1 v1.0.0-20170518105555-d52139073376

go.sum

@@ -572,8 +572,8 @@ github.com/jackc/pgproto3/v2 v2.3.3 h1:1HLSx5H+tXR9pW3in3zaztoEwQYRC9SQaYUHjTSUO
 github.com/jackc/pgproto3/v2 v2.3.3/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.5.5 h1:amBjrZVmksIdNjxGW/IiIMzxMKZFelXbUoPNb+8sjQw=
-github.com/jackc/pgx/v5 v5.5.5/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
+github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
+github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
 github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
 github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
@@ -813,8 +813,8 @@ github.com/turbot/go-kit v0.10.0-rc.0 h1:kd+jp2ibbIV33Hc8SsMAN410Dl9Pz6SJ40axbKU
 github.com/turbot/go-kit v0.10.0-rc.0/go.mod h1:fFQqR59I5z5JeeBLfK1PjSifn4Oprs3NiQx0CxeSJxs=
 github.com/turbot/go-prompt v0.2.6-steampipe.0.0.20221028122246-eb118ec58d50 h1:zs87uA6QZsYLk4RRxDOIxt8ro/B2V6HzoMWm05Lo7ao=
 github.com/turbot/go-prompt v0.2.6-steampipe.0.0.20221028122246-eb118ec58d50/go.mod h1:vFnjEGDIIA/Lib7giyE4E9c50Lvl8j0S+7FVlAwDAVw=
-github.com/turbot/pipe-fittings v1.2.0 h1:7JcSlmguYZvma+sjVCf74Nyv+oZcV++eGnVOQLzDSyU=
-github.com/turbot/pipe-fittings v1.2.0/go.mod h1:6q41X0QF4c2QjaORTvJEqXjTbp0Wp+Dx8cVLNXJQT60=
+github.com/turbot/pipe-fittings v1.5.2 h1:0nAIJD//lyxRVGJltRg+sCq98U/N1mGxu4LNWUJSPX8=
+github.com/turbot/pipe-fittings v1.5.2/go.mod h1:IgGxXEXfG3UTJCIkDOjjErVzU1ClsK6NmxlyaapZaNc=
 github.com/turbot/steampipe-cloud-sdk-go v0.6.0 h1:ufAxOpKS1uq7eejuE5sfEu1+d7QAd0RBjl8Bn6+mIs8=
 github.com/turbot/steampipe-cloud-sdk-go v0.6.0/go.mod h1:M42TMBdMim4bV1YTMxhKyzfSGSMo4CXUkm3wt9w7t1Y=
 github.com/turbot/steampipe-plugin-sdk/v5 v5.10.3 h1:3OCvnD0/u2RzNOgd0aBiWJrp/uk5Qjv6ZUt2gF2Xi9g=
@@ -1038,8 +1038,8 @@ golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220929204114-8fcdb60fdcc0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
-golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
+golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -1153,8 +1153,8 @@ golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
-golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
-golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
+golang.org/x/text v0.17.0 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc=
+golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

pkg/connection/refresh_connections_state.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -35,8 +36,11 @@ type connectionError struct {
 
 type refreshConnectionState struct {
 	// a connection pool to the DB service which uses the server appname
-	pool                       *pgxpool.Pool
-	searchPath                 []string
+	pool *pgxpool.Pool
+	// connectionOrder is the order of connections to be updated
+	// it is the search path, with any connections NOT in the searfch path in alphabetical order at the end
+	connectionOrder []string
+
 	connectionUpdates          *steampipeconfig.ConnectionUpdates
 	tableUpdater               *connectionStateTableUpdater
 	res                        *steampipeconfig.RefreshConnectionResult
@@ -64,9 +68,16 @@ func newRefreshConnectionState(ctx context.Context, pluginManager pluginManager,
 		return nil, err
 	}
 
+	//build list of connections in search path order, (with non search path connections at the end)
+	// get connections which are not in the search path
+	nonSearchPathConnections := steampipeconfig.GlobalConfig.GetNonSearchPathConnections(searchPath)
+	// sort alphabetically
+	slices.Sort(nonSearchPathConnections)
+	connectionOrder := append(searchPath, nonSearchPathConnections...)
+
 	res := &refreshConnectionState{
 		pool:                       pool,
-		searchPath:                 searchPath,
+		connectionOrder:            connectionOrder,
 		forceUpdateConnectionNames: forceUpdateConnectionNames,
 		pluginManager:              pluginManager,
 	}
@@ -151,6 +162,7 @@ func (s *refreshConnectionState) refreshConnections(ctx context.Context) {
 
 	// NOTE: delete any DYNAMIC plugin connections which will be updated
 	// to avoid them being accessed before they are updated
+	log.Printf("[TRACE] deleting %d dynamic plugin connections to avoid them being accessed before they are updated", len(s.connectionUpdates.DynamicUpdates()))
 	if err := s.executeDeleteQueries(ctx, s.connectionUpdates.DynamicUpdates()); err != nil {
 		s.res.Error = err
 		return
@@ -373,16 +385,20 @@ func (s *refreshConnectionState) executeUpdateQueries(ctx context.Context) {
 	log.Printf("[INFO] executing %d update %s", numUpdates, utils.Pluralize("query", numUpdates))
 
 	// execute initial updates
-	log.Printf("[INFO] executing initial updates")
 	var errors []error
-	moreErrors := s.executeUpdatesInParallel(ctx, initialUpdates)
-	errors = append(errors, moreErrors...)
+	if len(initialUpdates) > 0 {
+		log.Printf("[INFO] executing %d initial %s", len(initialUpdates), utils.Pluralize("update", len(initialUpdates)))
+		moreErrors := s.executeUpdatesInParallel(ctx, initialUpdates)
+		errors = append(errors, moreErrors...)
+	}
 
-	// execute dynamic updates (note, we update all connections in search path order,
-	// so must call executeUpdateSetsInParallel)
-	log.Printf("[INFO] executing dynamic updates")
-	moreErrors = s.executeUpdateSetsInParallel(ctx, dynamicUpdates)
-	errors = append(errors, moreErrors...)
+	if len(dynamicUpdates) > 0 {
+		// execute dynamic updates (note, we update all connections in search path order,
+		// so must call executeUpdateSetsInParallel)
+		log.Printf("[INFO] executing %d dynamic %s", len(dynamicUpdates), utils.Pluralize("update", len(dynamicUpdates)))
+		moreErrors := s.executeUpdateSetsInParallel(ctx, dynamicUpdates)
+		errors = append(errors, moreErrors...)
+	}
 
 	// if any of the initial schemas failed, do not proceed - these schemas are required to ensure we correctly
 	// resolve unqualified queries/tables
@@ -410,12 +426,14 @@ func (s *refreshConnectionState) executeUpdateQueries(ctx context.Context) {
 		log.Printf("[WARN] failed to send schem update Postgres notification: %s", err.Error())
 	}
 
-	log.Printf("[INFO] Execute %d remaining %s",
-		len(remainingUpdates),
-		utils.Pluralize("updates", len(remainingUpdates)))
-	// now execute remaining updates
-	moreErrors = s.executeUpdatesInParallel(ctx, remainingUpdates)
-	errors = append(errors, moreErrors...)
+	if len(remainingUpdates) > 0 {
+		log.Printf("[INFO] Execute %d remaining %s",
+			len(remainingUpdates),
+			utils.Pluralize("updates", len(remainingUpdates)))
+		// now execute remaining updates
+		moreErrors := s.executeUpdatesInParallel(ctx, remainingUpdates)
+		errors = append(errors, moreErrors...)
+	}
 
 	log.Printf("[INFO] Set comments for %d remaining %s and %d %s missing comments",
 		len(remainingUpdates),
@@ -520,7 +538,7 @@ func (s *refreshConnectionState) executeUpdateSetsInParallel(ctx context.Context
 	if envClone, ok := os.LookupEnv("STEAMPIPE_CLONE_SCHEMA"); ok {
 		cloneSchemaEnabled = strings.ToLower(envClone) == "true"
 	}
-	log.Printf("[INFO] executeUpdateForConnections - cloneSchema=%v", cloneSchemaEnabled)
+	log.Printf("[INFO] executeUpdateForConnections - cloneSchemaEnabled=%v", cloneSchemaEnabled)
 
 	// each update may be multiple connections, to execute in order
 	for _, states := range updates {
@@ -649,7 +667,6 @@ func (s *refreshConnectionState) UpdateCommentsInParallel(ctx context.Context, u
 					return
 				}
 				errors = append(errors, connectionError.err)
-				// TODO just log errors
 			}
 		}
 	}()
@@ -682,6 +699,8 @@ func (s *refreshConnectionState) UpdateCommentsInParallel(ctx context.Context, u
 
 // syncronously execute the comments queries for one or more connections
 func (s *refreshConnectionState) updateCommentsForConnection(ctx context.Context, errChan chan *connectionError, connectionPluginMap map[string]*steampipeconfig.ConnectionPlugin, connectionState *steampipeconfig.ConnectionState) {
+	log.Printf("[DEBUG] refreshConnectionState.updateCommentsForConnection start for connection '%s'", connectionState.ConnectionName)
+
 	connectionName := connectionState.ConnectionName
 
 	var sql string
@@ -771,7 +790,7 @@ func getCloneSchemaQuery(exemplarSchemaName string, connectionState *steampipeco
 
 func (s *refreshConnectionState) getInitialAndRemainingUpdates() (initialUpdates, remainingUpdates map[string]*steampipeconfig.ConnectionState, dynamicUpdates map[string][]*steampipeconfig.ConnectionState) {
 	updates := s.connectionUpdates.Update
-	searchPathConnections := s.connectionUpdates.FinalConnectionState.GetFirstSearchPathConnectionForPlugins(s.searchPath)
+	searchPathConnections := s.connectionUpdates.FinalConnectionState.GetFirstSearchPathConnectionForPlugins(s.connectionOrder)
 
 	initialUpdates = make(map[string]*steampipeconfig.ConnectionState)
 	remainingUpdates = make(map[string]*steampipeconfig.ConnectionState)
@@ -796,8 +815,20 @@ func (s *refreshConnectionState) getInitialAndRemainingUpdates() (initialUpdates
 		if connectionState.SchemaMode == plugin.SchemaModeStatic && !isInitialUpdate {
 			remainingUpdates[connectionName] = connectionState
 		}
+	}
 
+	log.Printf("[TRACE] getInitialAndRemainingUpdates: %d initialUpdates: %s, %d remainingUpdates: %s, %d dynamicUpdates: %s",
+		len(initialUpdates),
+		strings.Join(maps.Keys(initialUpdates), ", "),
+		len(remainingUpdates),
+		strings.Join(maps.Keys(remainingUpdates), ", "),
+		len(dynamicUpdates),
+		strings.Join(maps.Keys(dynamicUpdates), ", "))
+
+	if len(initialUpdates)+len(dynamicUpdates)+len(remainingUpdates) != len(updates) {
+		log.Printf("[WARN] getInitialAndRemainingUpdates: initialUpdates + remainingUpdates + dynamicUpdates != updates")
 	}
+
 	return initialUpdates, remainingUpdates, dynamicUpdates
 }
 

pkg/steampipeconfig/connection_updates.go

@@ -423,6 +423,7 @@ func (u *ConnectionUpdates) IdentifyMissingComments() {
 				_, updating := u.Update[name]
 				_, deleting := u.Delete[name]
 				if !updating || deleting {
+					log.Printf("[TRACE] connection %s comments not set, marking as missing", name)
 					u.MissingComments[name] = state
 				}
 			}

pkg/steampipeconfig/steampipeconfig.go

@@ -17,6 +17,7 @@ import (
 	"github.com/turbot/steampipe/pkg/ociinstaller/versionfile"
 	"github.com/turbot/steampipe/pkg/steampipeconfig/modconfig"
 	"github.com/turbot/steampipe/pkg/steampipeconfig/options"
+	"github.com/turbot/steampipe/pkg/utils"
 )
 
 // SteampipeConfig is a struct to hold Connection map and Steampipe options
@@ -306,8 +307,8 @@ func (c *SteampipeConfig) addPlugin(plugin *modconfig.Plugin) error {
 		log.Printf("[WARN] addPlugin called for plugin '%s' which is not installed", imageRef)
 		return nil
 	}
-		//  populate the version from the plugin version file data
-		plugin.Version = pluginVersion.Version
+	//  populate the version from the plugin version file data
+	plugin.Version = pluginVersion.Version
 
 	// add to list of plugin configs for this image ref
 	c.Plugins[imageRef] = append(c.Plugins[imageRef], plugin)
@@ -442,3 +443,17 @@ func (c *SteampipeConfig) resolvePluginInstanceForConnection(connection *modconf
 		return nil, sperr.New("connection '%s' specifies 'plugin=\"%s\"' but the correct instance cannot be uniquely resolved. There are %d plugin instances matching that configuration:\n%s", connection.Name, connection.PluginAlias, len(pluginsForImageRef), strings.Join(strs, "\n"))
 	}
 }
+
+// GetNonSearchPathConnections returns a list of connection names that are not in the provided search path
+func (c *SteampipeConfig) GetNonSearchPathConnections(searchPath []string) []string {
+	var res []string
+	//convert searchPath to map for easy lookup
+	searchPathLookup := utils.SliceToLookup(searchPath)
+
+	for connectionName, _ := range c.Connections {
+		if _, inSearchPath := searchPathLookup[connectionName]; !inSearchPath {
+			res = append(res, connectionName)
+		}
+	}
+	return res
+}

pkg/version/version.go

@@ -18,7 +18,7 @@
 **/
 
 // The main version number that is being run at the moment.
-var steampipeVersion = "0.23.4"
+var steampipeVersion = "0.23.5"
 
 // A pre-release marker for the version. If this is "" (empty string)
 // then it means that it is a final release. Otherwise, this is a pre-release
@@ -28,7 +28,7 @@
 // SteampipeVersion is an instance of semver.Version. This has the secondary
 // benefit of verifying during tests and init time that our version is a
 // proper semantic version, which should always be the case.
 var SteampipeVersion *semver.Version
 
 var VersionString string
 
@@ -37,5 +37,5 @@
 	if prerelease != "" {
 		VersionString = fmt.Sprintf("%s-%s", steampipeVersion, prerelease)
 	}
 	SteampipeVersion = semver.MustParse(VersionString)
 }

tests/acceptance/test_data/source_files/default_search_path.spc

@@ -0,0 +1,3 @@
+options "database" {
+  search_path        = "public,chaos"
+}

tests/acceptance/test_data/source_files/servicenow.spc

@@ -0,0 +1,6 @@
+connection "new_servicenow" {
+  plugin = "servicenow"
+  instance_url = "https://fakeinstance11.service-now.com"
+  username = "fakeuser11"
+  password = "fakepassword11"
+}

tests/acceptance/test_files/connection_config.bats

@@ -194,7 +194,32 @@ load "$LIB_BATS_SUPPORT/load.bash"
     assert_success
 }
 
+# This test checks this pipes bug - https://github.com/turbot/steampipe/issues/4353
+# With service running, if a new connection is added but is not in search_path, it should be available and ready
+# previously, the connection was in error state
+# NOTE: This test should always be the last test in this file
+@test "dynamic schema - service running, new connection added(but not in search_path) - connection should be available and ready" {
+  steampipe plugin install servicenow --install-dir $STEAMPIPE_INSTALL_DIR
+  # start service
+  steampipe service start --install-dir $STEAMPIPE_INSTALL_DIR
+
+  # update search_path in db options, to exclude the new connection
+  cp $SRC_DATA_DIR/default_search_path.spc $STEAMPIPE_INSTALL_DIR/config/default.spc
+
+	cat $STEAMPIPE_INSTALL_DIR/config/default.spc
+
+  # add a new connection
+  cp $SRC_DATA_DIR/servicenow.spc $STEAMPIPE_INSTALL_DIR/config/servicenow2.spc
+
+  sleep 10
+
+  # check if the new connection is available and ready
+  run steampipe query "select name, state from steampipe_connection" --output csv --install-dir $STEAMPIPE_INSTALL_DIR
+  assert_output --partial 'servicenow,ready'
+}
+
 @test "cleanup" {
+  steampipe service stop --install-dir $STEAMPIPE_INSTALL_DIR
   rm -f $STEAMPIPE_INSTALL_DIR/config/chaos_agg.spc
   run steampipe plugin uninstall steampipe
   rm -f $STEAMPIPE_INSTALL_DIR/config/steampipe.spc