feat: init steampipe

go.mod

@@ -43,7 +43,7 @@ require (
 	github.com/lib/pq v1.10.9
 	github.com/nats-io/nats.go v1.36.0
 	github.com/open-policy-agent/opa v0.69.0
-	github.com/opengovern/og-util v1.1.13
+	github.com/opengovern/og-util v1.1.14-0.20241219130558-d70ee92c91fa
 	github.com/opensearch-project/opensearch-go/v2 v2.3.0
 	github.com/opensearch-project/opensearch-go/v4 v4.2.0
 	github.com/ory/dockertest/v3 v3.10.0
@@ -199,14 +199,10 @@ require (
 	github.com/iancoleman/strcase v0.3.0 // indirect
 	github.com/imdario/mergo v0.3.16 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
-	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
-	github.com/jackc/pgconn v1.14.3 // indirect
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
-	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/pgx/v4 v4.18.3 // indirect
-	github.com/jackc/puddle v1.3.0 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect

go.sum

@@ -371,7 +371,6 @@ github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWH
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20240723142845-024c85f92f20 h1:N+3sFI5GUjRKBi+i0TxYVST9h4Ie192jJWpHvthBBgg=
 github.com/cncf/xds/go v0.0.0-20240723142845-024c85f92f20/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
-github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I=
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
 github.com/confluentinc/confluent-kafka-go/v2 v2.3.0 h1:icCHutJouWlQREayFwCc7lxDAhws08td+W3/gdqgZts=
 github.com/confluentinc/confluent-kafka-go/v2 v2.3.0/go.mod h1:/VTy8iEpe6mD9pkCH5BhijlUl8ulUXymKv1Qig5Rgb8=
@@ -574,8 +573,6 @@ github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJA
 github.com/goccy/go-yaml v1.11.2 h1:joq77SxuyIs9zzxEjgyLBugMQ9NEgTWxXfz2wVqwAaQ=
 github.com/goccy/go-yaml v1.11.2/go.mod h1:wKnAMd44+9JAAnGQpWVEgBzGt3YuTaQ4uXoHvE4m7WU=
 github.com/gofrs/uuid v4.0.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
-github.com/gofrs/uuid v4.4.0+incompatible h1:3qXRTX8/NbyulANqlc0lchS1gqAVxRgsuW1YrTJupqA=
-github.com/gofrs/uuid v4.4.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/gogo/googleapis v1.4.1 h1:1Yx4Myt7BxzvUr5ldGSbwYiZG6t9wGBZ+8/fX3Wvtq0=
 github.com/gogo/googleapis v1.4.1/go.mod h1:2lpHqI5OcWCtVElxXnPt+s8oJvMpySlOyM6xDCrzib4=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -765,6 +762,7 @@ github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4=
 github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jackc/chunkreader v1.0.0 h1:4s39bBR8ByfqH+DKm8rQA3E1LHZWB9XWcrz8fqaZbe0=
 github.com/jackc/chunkreader v1.0.0/go.mod h1:RT6O25fNZIuasFJRyZ4R/Y2BbhasbmZXF9QQ7T3kePo=
 github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
 github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8=
@@ -781,10 +779,10 @@ github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=
 github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8=
 github.com/jackc/pgmock v0.0.0-20190831213851-13a1b77aafa2/go.mod h1:fGZlG77KXmcq05nJLRkk0+p82V8B8Dw8KN2/V9c/OAE=
 github.com/jackc/pgmock v0.0.0-20201204152224-4fe30f7445fd/go.mod h1:hrBW0Enj2AZTNpt/7Y5rr2xe/9Mn757Wtb2xeBzPv2c=
-github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65 h1:DadwsjnMwFjfWc9y5Wi/+Zz7xoE5ALHsRQlOctkOiHc=
 github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65/go.mod h1:5R2h2EEX+qri8jOWMbJCtaPWkrrNc7OHwsp2TCqp7ak=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgproto3 v1.1.0 h1:FYYE4yRw+AgI8wXIinMlNjBbp/UitDJwfj5LqqewP1A=
 github.com/jackc/pgproto3 v1.1.0/go.mod h1:eR5FA3leWg7p9aeAqi37XOTgTIbkABlvcPB3E5rlc78=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190420180111-c116219b62db/go.mod h1:bhq50y+xrl9n5mRYyCBFKkpRVTLYJVWeCc+mEAI3yXA=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190609003834-432c2951c711/go.mod h1:uH0AWtUmuShn0bcesswc4aBTWGvw0cAxIJp+6OB//Wg=
@@ -814,8 +812,6 @@ github.com/jackc/pgx/v5 v5.7.1/go.mod h1:e7O26IywZZ+naJtWWos6i6fvWK+29etgITqrqHL
 github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
-github.com/jackc/puddle v1.3.0 h1:eHK/5clGOatcjX3oWGBO/MpxpbHzSwud5EWTSCI+MX0=
-github.com/jackc/puddle v1.3.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
@@ -1001,8 +997,8 @@ github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQ
 github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
 github.com/opencontainers/runc v1.2.0 h1:qke7ZVCmJcKrJVY2iHJVC+0kql9uYdkusOPsQOOeBw4=
 github.com/opencontainers/runc v1.2.0/go.mod h1:/PXzF0h531HTMsYQnmxXkBD7YaGShm/2zcRB79dksUc=
-github.com/opengovern/og-util v1.1.13 h1:WFL0hOaKtzpGLk5xGM8QeLs4Dn02JG0m7xIAgepo5yg=
-github.com/opengovern/og-util v1.1.13/go.mod h1:dyn8rhmxq59o1jnbgGfmcUvW7iB/eN6OxoTUUx6jEHA=
+github.com/opengovern/og-util v1.1.14-0.20241219130558-d70ee92c91fa h1:srcr2aeDLSN+t2BhuB9JvWTUDhvBx1DVqXhEXasSSV4=
+github.com/opengovern/og-util v1.1.14-0.20241219130558-d70ee92c91fa/go.mod h1:Q7Pd/1SzDtYoF3iAd4/FSBmFihis+7OIeDpU94xawXQ=
 github.com/opensearch-project/opensearch-go/v2 v2.3.0 h1:nQIEMr+A92CkhHrZgUhcfsrZjibvB3APXf2a1VwCmMQ=
 github.com/opensearch-project/opensearch-go/v2 v2.3.0/go.mod h1:8LDr9FCgUTVoT+5ESjc2+iaZuldqE+23Iq0r1XeNue8=
 github.com/opensearch-project/opensearch-go/v4 v4.2.0 h1:uaBexfVdeSU15yOUPYF+IY059koVP0oNQPyoSde6N/A=
@@ -1105,8 +1101,6 @@ github.com/sethvargo/go-retry v0.2.4 h1:T+jHEQy/zKJf5s95UkguisicE0zuF9y7+/vgz08O
 github.com/sethvargo/go-retry v0.2.4/go.mod h1:1afjQuvh7s4gflMObvjLPaWgluLLyhA1wmVZ6KLpICw=
 github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
 github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
-github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8=
-github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=

services/rego/cmd.go

@@ -1,13 +1,17 @@
 package rego
 
 import (
+	config2 "github.com/opengovern/og-util/pkg/config"
 	"github.com/opengovern/og-util/pkg/httpserver"
 	"github.com/opengovern/og-util/pkg/koanf"
+	"github.com/opengovern/og-util/pkg/steampipe"
+	integration_type "github.com/opengovern/opencomply/services/integration/integration-type"
 	"github.com/opengovern/opencomply/services/rego/api"
 	"github.com/opengovern/opencomply/services/rego/config"
 	"github.com/opengovern/opencomply/services/rego/internal"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
+	"time"
 )
 
 func Command() *cobra.Command {
@@ -24,7 +28,56 @@ func Command() *cobra.Command {
 
 			logger = logger.Named("rego")
 
-			regoEngine, err := internal.NewRegoEngine(ctx, logger)
+			for _, integrationType := range integration_type.IntegrationTypes {
+				describerConfig := integrationType.GetConfiguration()
+				err := steampipe.PopulateSteampipeConfig(config2.ElasticSearch{
+					Address:           cnf.ElasticSearch.Address,
+					Username:          cnf.ElasticSearch.Username,
+					Password:          cnf.ElasticSearch.Password,
+					IsOpenSearch:      cnf.ElasticSearch.IsOpenSearch,
+					IsOnAks:           cnf.ElasticSearch.IsOnAks,
+					AwsRegion:         cnf.ElasticSearch.AWSRegion,
+					AssumeRoleArn:     cnf.ElasticSearch.AssumeRoleARN,
+					ExternalID:        cnf.ElasticSearch.ExternalID,
+					IngestionEndpoint: cnf.ElasticSearch.IngestionEndpoint,
+				}, describerConfig.SteampipePluginName)
+				if err != nil {
+					return err
+				}
+			}
+			if err := steampipe.PopulateOpenGovernancePluginSteampipeConfig(config2.ElasticSearch{
+				Address:           cnf.ElasticSearch.Address,
+				Username:          cnf.ElasticSearch.Username,
+				Password:          cnf.ElasticSearch.Password,
+				IsOpenSearch:      cnf.ElasticSearch.IsOpenSearch,
+				IsOnAks:           cnf.ElasticSearch.IsOnAks,
+				AwsRegion:         cnf.ElasticSearch.AWSRegion,
+				AssumeRoleArn:     cnf.ElasticSearch.AssumeRoleARN,
+				ExternalID:        cnf.ElasticSearch.ExternalID,
+				IngestionEndpoint: cnf.ElasticSearch.IngestionEndpoint,
+			}, config2.Postgres{
+				Host:            cnf.Steampipe.Host,
+				Port:            cnf.Steampipe.Port,
+				DB:              cnf.Steampipe.DB,
+				Username:        cnf.Steampipe.Username,
+				Password:        cnf.Steampipe.Password,
+				SSLMode:         cnf.Steampipe.SSLMode,
+				MaxIdleConns:    cnf.Steampipe.MaxIdleConns,
+				MaxOpenConns:    cnf.Steampipe.MaxOpenConns,
+				ConnMaxIdleTime: cnf.Steampipe.ConnMaxIdleTime,
+				ConnMaxLifetime: cnf.Steampipe.ConnMaxLifetime,
+			}); err != nil {
+				return err
+			}
+
+			time.Sleep(2 * time.Minute)
+
+			steampipeConn, err := steampipe.StartSteampipeServiceAndGetConnection(logger)
+			if err != nil {
+				return err
+			}
+
+			regoEngine, err := internal.NewRegoEngine(ctx, logger, steampipeConn)
 			if err != nil {
 				return err
 			}

services/rego/config/service.go

@@ -3,5 +3,7 @@ package config
 import "github.com/opengovern/og-util/pkg/koanf"
 
 type RegoConfig struct {
-	Http koanf.HttpServer `json:"http,omitempty" koanf:"http"`
+	Http          koanf.HttpServer    `json:"http,omitempty" koanf:"http"`
+	ElasticSearch koanf.ElasticSearch `json:"elasticsearch,omitempty" koanf:"elasticsearch"`
+	Steampipe     koanf.Postgres      `json:"steampipe,omitempty" koanf:"steampipe"`
 }

services/rego/internal/rego.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/jackc/pgx/v5"
-	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/open-policy-agent/opa/ast"
 	"github.com/open-policy-agent/opa/rego"
 	"github.com/open-policy-agent/opa/types"
@@ -14,43 +13,21 @@ import (
 )
 
 type RegoEngine struct {
-	logger *zap.Logger
-	db     *pgxpool.Pool
+	logger    *zap.Logger
+	steampipe *steampipesdk.Database
 
 	regoFunctions []func(*rego.Rego)
 }
 
 var excludedTableSchema = []string{"information_schema", "pg_catalog", "steampipe_internal", "steampipe_command", "public"}
 
-func NewRegoEngine(ctx context.Context, logger *zap.Logger) (*RegoEngine, error) {
+func NewRegoEngine(ctx context.Context, logger *zap.Logger, steampipeDb *steampipesdk.Database) (*RegoEngine, error) {
 	engine := RegoEngine{
-		logger: logger,
+		logger:    logger,
+		steampipe: steampipeDb,
 	}
-	option := steampipesdk.GetDefaultSteampipeOption()
-	selfClientConfig, err := pgxpool.ParseConfig(fmt.Sprintf(`host=%s port=%s user=%s password=%s dbname=%s sslmode=disable TimeZone=GMT`, option.Host, option.Port, option.User, option.Pass, option.Db))
-	if err != nil {
-		logger.Error("Unable to parse config", zap.Error(err))
-		return nil, err
-	}
-	logger.Info("Connecting to database", zap.String("host", option.Host), zap.String("port", option.Port), zap.String("user", option.User), zap.String("db", option.Db))
 
 	tries := 5
-	for i := 0; i < tries; i++ {
-		db, err := pgxpool.NewWithConfig(ctx, selfClientConfig)
-		if err != nil {
-			logger.Error("Unable to connect to database", zap.Error(err), zap.Int("try", i))
-			if i == tries-1 {
-				logger.Error("Exhausted all tries to connect to database")
-				return nil, err
-			}
-			time.Sleep(10 * time.Second)
-			continue
-		}
-		engine.db = db
-		break
-	}
-	logger.Info("Connected to database")
-
 	for i := 0; i < tries; i++ {
 		functions, err := engine.getRegoFunctionForTables(ctx)
 		if err != nil {
@@ -72,7 +49,7 @@ func NewRegoEngine(ctx context.Context, logger *zap.Logger) (*RegoEngine, error)
 
 func (r *RegoEngine) getRegoFunctionForTables(ctx context.Context) ([]func(*rego.Rego), error) {
 
-	rows, err := r.db.Query(ctx, "SELECT table_name FROM information_schema.tables WHERE table_schema != any ($1)", excludedTableSchema)
+	rows, err := r.steampipe.Conn().Query(ctx, "SELECT table_name FROM information_schema.tables WHERE table_schema != any ($1)", excludedTableSchema)
 	if err != nil {
 		r.logger.Error("Unable to query database", zap.Error(err))
 		r.logger.Sync()
@@ -97,7 +74,7 @@ func (r *RegoEngine) getRegoFunctionForTables(ctx context.Context) ([]func(*rego
 			Memoize:          true,
 			Nondeterministic: true,
 		}, func(bctx rego.BuiltinContext, terms []*ast.Term) (*ast.Term, error) {
-			rows, err := r.db.Query(bctx.Context, fmt.Sprintf("SELECT * FROM %s", tableName))
+			rows, err := r.steampipe.Conn().Query(bctx.Context, fmt.Sprintf("SELECT * FROM %s", tableName))
 			if err != nil {
 				r.logger.Error("Unable to query database", zap.Error(err), zap.String("table", tableName))
 				r.logger.Sync()