Add support for running plugins in-process. Closes #383

Author: kaidaguerre

Dockerfile

@@ -2,19 +2,24 @@
 #
 # Dockerfile for cross-compiling for Linux on MacOS
 # Build the image with:
-#       docker build --pull -f Dockerfile -t steampipe_fdw_builder:13 .
+#   docker build --pull -f Dockerfile -t steampipe_fdw_builder:15 --build-arg="pg_version=15" .  
 #
 # Run with:
-#       docker run -it --rm --name sp_fdw_builder_13  -v $(pwd):/tmp/ext steampipe_fdw_builder:13
+#   docker run -it --rm --name sp_fdw_builder  -v $(pwd):/tmp/ext steampipe_fdw_builder:15
 #
 #####
 
 FROM ubuntu:focal
 
+# We know that the FDW does not compile with PG12. 
+# Use this so that the build fails if an ARG is not passed in.
+# This is useful since we can use the same container definition for the SQLite builder as well
+ARG pg_version=12
 ARG go_repo="deb http://ppa.launchpad.net/longsleep/golang-backports/ubuntu bionic main"
 ARG pg_repo="deb http://apt.postgresql.org/pub/repos/apt/ focal-pgdg main"
-ENV PG_VERS=13
-ENV GO_VERS=1.16
+
+ENV PG_VERS=$pg_version
+ENV GO_VERS=1.21
 
 ## for apt to be noninteractive
 ARG DEBIAN_FRONTEND=noninteractive
@@ -28,25 +33,25 @@ RUN apt-get install -y --no-install-recommends curl
 RUN apt-get install -y --no-install-recommends ca-certificates
 
 RUN mkdir -p /etc/apt/sources.list.d \
-    && apt-key adv --keyserver keyserver.ubuntu.com --recv 56A3D45E \
-    && apt-key adv --keyserver keyserver.ubuntu.com --recv E0C56BD4 \
-    && echo $go_repo > /etc/apt/sources.list.d/golang.list \
-    && echo $pg_repo > /etc/apt/sources.list.d/pgdb.list \
-    && curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add -
- 
+  && apt-key adv --keyserver keyserver.ubuntu.com --recv 56A3D45E \
+  && apt-key adv --keyserver keyserver.ubuntu.com --recv E0C56BD4 \
+  && echo $go_repo > /etc/apt/sources.list.d/golang.list \
+  && echo $pg_repo > /etc/apt/sources.list.d/pgdb.list \
+  && curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add -
+
 RUN apt-get update
 RUN env DEBIAN_FRONTEND=noninteractive \
-        apt-get install -y --no-install-recommends golang-${GO_VERS} \
-            postgresql-${PG_VERS} postgresql-server-dev-${PG_VERS} libpq-dev wget build-essential \
-            libgcc-7-dev \
-            locales \
-            tzdata \
-            git \
-    && rm -rf \
-        /var/lib/apt/lists/* \
-        /var/cache/debconf \
-        /tmp/* \
-    && apt-get clean
+  apt-get install -y --no-install-recommends golang-${GO_VERS} \
+  postgresql-${PG_VERS} postgresql-server-dev-${PG_VERS} libpq-dev wget build-essential \
+  libgcc-7-dev \
+  locales \
+  tzdata \
+  git \
+  && rm -rf \
+  /var/lib/apt/lists/* \
+  /var/cache/debconf \
+  /tmp/* \
+  && apt-get clean
 
 RUN ln -s /usr/lib/go-${GO_VERS}/bin/go /usr/bin/go
 RUN locale-gen en_US.UTF-8

LICENSE

@@ -1,6 +1,7 @@
 # Makefile
 
 PLATFORM=$(shell uname)
+GETTEXT_INCLUDE=$(shell dirname $(shell dirname $(shell readlink -f $(shell which gettext))))/include
 
 install: build
 	if test -d ~/.steampipe/db/14.2.0; then \
@@ -9,6 +10,23 @@ install: build
 		cp ./build-$(PLATFORM)/steampipe_postgres_fdw.so ~/.steampipe/db/14.2.0/postgres/lib/postgresql/; \
 	fi
 
+# build standalone 
+standalone: validate_plugin prebuild.go
+	@echo "Building standalone FDW for plugin: $(plugin)"
+	go run generate/generator.go templates . $(plugin) $(plugin_github_url)
+	go mod tidy
+	$(MAKE) -C ./fdw clean
+	$(MAKE) -C ./fdw go
+	$(MAKE) -C ./fdw
+	$(MAKE) -C ./fdw standalone
+	
+	rm -f prebuild.go
+
+validate_plugin:
+    ifndef plugin
+	    $(error "You must specify the 'plugin' variable")
+    endif
+
 build: prebuild.go
 	$(MAKE) -C ./fdw clean
 	$(MAKE) -C ./fdw go
@@ -31,6 +49,7 @@ prebuild.go:
 	sed -i.bak 's|INTERNAL_INCLUDE_PLACEHOLDER|$(shell pg_config --includedir)|' prebuild.go
 	sed -i.bak 's|SERVER_INCLUDE_PLACEHOLDER|$(shell pg_config --includedir-server)|' prebuild.go
 	sed -i.bak 's|DISCLAIMER|This is generated. Do not check this in to Git|' prebuild.go
+	sed -i.bak 's|LIB_INTL_PLACEHOLDER|$(GETTEXT_INCLUDE)|' prebuild.go
 	rm -f prebuild.go.bak
 
 clean:
@@ -43,3 +62,4 @@ clean:
 # Usage: make release input="v1.7.2"
 release:
 	./scripts/upload_arm_asset.sh $(input)
+

Makefile

@@ -54,6 +54,33 @@ $ make
 
 This will compile the FDW (`steampipe_postgres_fdw.so`) along with the `control` and `sql` file in the `build-$PLATFORM` directory. This will install the compiled FDW into the default Steampipe installation directory (`~/.steampipe`) - if it exists.
 
+### Building the FDW for a Single Plugin
+
+If you want to build the FDW for a single steampipe plugin, follow these steps. This process allows you to build the Postgres Foreign Data Wrapper (FDW) specifically for one particular plugin and run it in standalone mode on any PostgreSQL database, without any reliance on Steampipe.
+
+Make sure that you have the following installed in your system:
+1. `Postgresql v14` 
+1. `go`
+1. `gcc` for Linux
+
+Steps:
+1. Clone this repository onto your system
+1. Change to the cloned directory
+1. Run the following commands:
+```
+$ make standalone plugin="<plugin alias>"
+```
+Replace plugin alias with the alias or short name of your plugin.
+
+This command will compile the FDW specifically for the chosen plugin, and the resulting binary, control file, and SQL files will be generated.
+
+#### Example:
+
+Suppose you want to build the FDW for a plugin with an alias `aws` from a GitHub repository located at https://github.com/turbot/steampipe-plugin-aws. You would run the following command:
+```
+$ make standalone plugin="aws"
+```
+
 ### License
 
 This open source library is licensed under the [GNU Affero General Public License v3](https://opensource.org/licenses/AGPL-3.0).

README.md

@@ -13,14 +13,12 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
-	"net"
-	"net/http"
-	"os"
 	"time"
 	"unsafe"
 
 	"github.com/hashicorp/go-hclog"
 	"github.com/turbot/steampipe-plugin-sdk/v5/logging"
+	"github.com/turbot/steampipe-plugin-sdk/v5/sperr"
 	"github.com/turbot/steampipe-postgres-fdw/hub"
 	"github.com/turbot/steampipe-postgres-fdw/types"
 	"github.com/turbot/steampipe-postgres-fdw/version"
@@ -56,22 +54,14 @@ func init() {
 	log.SetOutput(logger.StandardWriter(&hclog.StandardLoggerOptions{InferLevels: true}))
 	log.SetPrefix("")
 	log.SetFlags(0)
+	// create hub
+	if err := hub.CreateHub(); err != nil {
+		panic(err)
+	}
 	log.Printf("[INFO] .\n******************************************************\n\n\t\tsteampipe postgres fdw init\n\n******************************************************\n")
 	log.Printf("[INFO] Version:   v%s\n", version.FdwVersion.String())
 	log.Printf("[INFO] Log level: %s\n", level)
 
-	if _, found := os.LookupEnv("STEAMPIPE_FDW_PPROF"); found {
-		log.Printf("[INFO] PROFILING!!!!")
-		go func() {
-			listener, err := net.Listen("tcp", "localhost:0")
-			if err != nil {
-				log.Println(err)
-				return
-			}
-			log.Printf("[INFO] Check http://localhost:%d/debug/pprof/", listener.Addr().(*net.TCPAddr).Port)
-			log.Println(http.Serve(listener, nil))
-		}()
-	}
 }
 
 //export goFdwGetRelSize
@@ -80,10 +70,17 @@ func goFdwGetRelSize(state *C.FdwPlanState, root *C.PlannerInfo, rows *C.double,
 
 	log.Printf("[TRACE] goFdwGetRelSize")
 
-	pluginHub, err := hub.GetHub()
+	pluginHub := hub.GetHub()
+
+	// get connection name
+	connName := GetSchemaNameFromForeignTableId(types.Oid(state.foreigntableid))
+
+	log.Println("[TRACE] connection name:", connName)
+
+	serverOpts := GetForeignServerOptionsFromFTableId(types.Oid(state.foreigntableid))
+	err := pluginHub.ProcessImportForeignSchemaOptions(serverOpts, connName)
 	if err != nil {
-		FdwError(err)
-		return
+		FdwError(sperr.WrapWithMessage(err, "failed to process options"))
 	}
 
 	// reload connection config
@@ -95,15 +92,15 @@ func goFdwGetRelSize(state *C.FdwPlanState, root *C.PlannerInfo, rows *C.double,
 		return
 	}
 
-	opts := GetFTableOptions(types.Oid(state.foreigntableid))
+	tableOpts := GetFTableOptions(types.Oid(state.foreigntableid))
 
 	// build columns
 	var columns []string
 	if state.target_list != nil {
 		columns = CStringListToGoArray(state.target_list)
 	}
 
-	result, err := pluginHub.GetRelSize(columns, nil, opts)
+	result, err := pluginHub.GetRelSize(columns, nil, tableOpts)
 	if err != nil {
 		log.Println("[ERROR] pluginHub.GetRelSize")
 		FdwError(err)
@@ -127,17 +124,12 @@ func goFdwGetPathKeys(state *C.FdwPlanState) *C.List {
 	}()
 
 	log.Printf("[TRACE] goFdwGetPathKeys")
-	pluginHub, err := hub.GetHub()
-	if err != nil {
-		FdwError(err)
-	}
+	pluginHub := hub.GetHub()
+
 	var result *C.List
 	opts := GetFTableOptions(types.Oid(state.foreigntableid))
-	ftable := C.GetForeignTable(state.foreigntableid)
-	rel := C.RelationIdGetRelation(ftable.relid)
-	defer C.RelationClose(rel)
 	// get the connection name - this is the namespace (i.e. the local schema)
-	opts["connection"] = getNamespace(rel)
+	opts["connection"] = GetSchemaNameFromForeignTableId(types.Oid(state.foreigntableid))
 
 	if opts["connection"] == constants.InternalSchema || opts["connection"] == constants.LegacyCommandSchema {
 		return result
@@ -240,12 +232,8 @@ func goFdwBeginForeignScan(node *C.ForeignScanState, eflags C.int) {
 	quals, unhandledRestrictions := restrictionsToQuals(node, cinfos)
 
 	// start the plugin hub
-	var err error
-	pluginHub, err := hub.GetHub()
-	if err != nil {
-		FdwError(err)
-	}
 
+	pluginHub := hub.GetHub()
 	s := &ExecState{
 		Rel:   rel,
 		Opts:  opts,
@@ -282,7 +270,7 @@ func goFdwIterateForeignScan(node *C.ForeignScanState) *C.TupleTableSlot {
 
 	slot := node.ss.ss_ScanTupleSlot
 	C.ExecClearTuple(slot)
-	pluginHub, _ := hub.GetHub()
+	pluginHub := hub.GetHub()
 
 	log.Printf("[TRACE] goFdwIterateForeignScan, table '%s' (%p)", s.Opts["table"], s.Iter)
 	// if the iterator has not started, start
@@ -367,8 +355,8 @@ func goFdwEndForeignScan(node *C.ForeignScanState) {
 		}
 	}()
 	s := GetExecState(node.fdw_state)
-	pluginHub, _ := hub.GetHub()
-	if s != nil && pluginHub != nil {
+	pluginHub := hub.GetHub()
+	if s != nil {
 		log.Printf("[INFO] goFdwEndForeignScan, iterator: %p", s.Iter)
 		pluginHub.EndScan(s.Iter, int64(s.State.limit))
 	}
@@ -386,9 +374,9 @@ func goFdwAbortCallback() {
 		}
 	}()
 	log.Printf("[INFO] goFdwAbortCallback")
-	if pluginHub, err := hub.GetHub(); err == nil {
-		pluginHub.Abort()
-	}
+	pluginHub := hub.GetHub()
+	pluginHub.Abort()
+
 }
 
 //export goFdwImportForeignSchema
@@ -402,12 +390,7 @@ func goFdwImportForeignSchema(stmt *C.ImportForeignSchemaStmt, serverOid C.Oid)
 
 	log.Printf("[INFO] goFdwImportForeignSchema remote '%s' local '%s'\n", C.GoString(stmt.remote_schema), C.GoString(stmt.local_schema))
 	// get the plugin hub,
-	pluginHub, err := hub.GetHub()
-	if err != nil {
-		log.Printf("[WARN] goFdwImportForeignSchema failed: %s", err)
-		FdwError(err)
-		return nil
-	}
+	pluginHub := hub.GetHub()
 
 	remoteSchema := C.GoString(stmt.remote_schema)
 	localSchema := C.GoString(stmt.local_schema)
@@ -426,6 +409,16 @@ func goFdwImportForeignSchema(stmt *C.ImportForeignSchemaStmt, serverOid C.Oid)
 		return sql
 	}
 
+	fServer := C.GetForeignServer(serverOid)
+	serverOptions := GetForeignServerOptions(fServer)
+
+	log.Println("[TRACE] goFdwImportForeignSchema serverOptions:", serverOptions)
+
+	err := pluginHub.ProcessImportForeignSchemaOptions(serverOptions, localSchema)
+	if err != nil {
+		FdwError(sperr.WrapWithMessage(err, "failed to process options"))
+	}
+
 	schema, err := pluginHub.GetSchema(remoteSchema, localSchema)
 	if err != nil {
 		log.Printf("[WARN] goFdwImportForeignSchema failed: %s", err)
@@ -462,31 +455,22 @@ func goFdwExecForeignInsert(estate *C.EState, rinfo *C.ResultRelInfo, slot *C.Tu
 func handleCommandInsert(rinfo *C.ResultRelInfo, slot *C.TupleTableSlot, rel C.Relation) *C.TupleTableSlot {
 	relid := rinfo.ri_RelationDesc.rd_id
 	opts := GetFTableOptions(types.Oid(relid))
+	pluginHub := hub.GetHub()
 
 	switch opts["table"] {
 	case constants.LegacyCommandTableCache:
 		// we know there is just a single column - operation
 		var isNull C.bool
 		datum := C.slot_getattr(slot, 1, &isNull)
 		operation := C.GoString(C.fdw_datumGetString(datum))
-		hub, err := hub.GetHub()
-		if err != nil {
-			FdwError(err)
-			return nil
-		}
-		if err := hub.HandleLegacyCacheCommand(operation); err != nil {
+		if err := pluginHub.HandleLegacyCacheCommand(operation); err != nil {
 			FdwError(err)
 			return nil
 		}
 
 	case constants.ForeignTableSettings:
 		tupleDesc := buildTupleDesc(rel.rd_att)
 		attributes := tupleDesc.Attrs
-		hub, err := hub.GetHub()
-		if err != nil {
-			FdwError(err)
-			return nil
-		}
 		var key *string
 		var value *string
 
@@ -520,7 +504,7 @@ func handleCommandInsert(rinfo *C.ResultRelInfo, slot *C.TupleTableSlot, rel C.R
 		}
 
 		// apply the setting
-		if err = hub.ApplySetting(*key, *value); err != nil {
+		if err := pluginHub.ApplySetting(*key, *value); err != nil {
 			FdwError(err)
 		}
 		return nil
@@ -548,10 +532,7 @@ func goFdwShutdown() {
 		}
 	}()
 	log.Printf("[INFO] .\n******************************************************\n\n\t\tsteampipe postgres fdw shutdown\n\n******************************************************\n")
-	pluginHub, err := hub.GetHub()
-	if err != nil {
-		FdwError(err)
-	}
+	pluginHub := hub.GetHub()
 	pluginHub.Close()
 }