Allow syncing via the connect experimental transport.

internal/connector/connector_connect.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap"
 	"golang.org/x/net/http2/h2c"
 
 	"golang.org/x/net/http2"
@@ -58,6 +59,7 @@ func NewWrapperConnect(ctx context.Context, server interface{}, optfunc ...Optio
 }
 
 func (cw *wrapperConnect) Run(ctx context.Context, serverCfg *connectorwrapperV1.ServerConfig) error {
+	ctxzap.Extract(ctx).Warn("connect enabled!")
 	rl, err := ratelimit2.NewLimiter(ctx, cw.now, serverCfg.RateLimiterConfig)
 	if err != nil {
 		return err

internal/connector/connector_connect_shim.go

@@ -31,7 +31,8 @@ func NewConnectorConnectShim(
 	ticketingEnabled bool,
 ) *connectorConnectShim {
 	s := &connectorConnectShim{
-		server: server,
+		server:      server,
+		rateLimiter: rateLimiter,
 	}
 
 	if ticketingEnabled {

pkg/cli/commands.go

@@ -166,6 +166,10 @@ func MakeMainCommand(
 			opts = append(opts, connectorrunner.WithTempDir(v.GetString("c1z-temp-dir")))
 		}
 
+		if v.GetBool("use-connect-experiment") {
+			opts = append(opts, connectorrunner.WithConnectExperiment())
+		}
+
 		// NOTE(shackra): top-most in the execution flow for connectors
 		r, err := connectorrunner.NewConnectorRunner(runCtx, c, opts...)
 		if err != nil {

pkg/connectorrunner/runner.go

@@ -299,6 +299,8 @@ type runnerConfig struct {
 	listTicketSchemasConfig *listTicketSchemasConfig
 	getTicketConfig         *getTicketConfig
 	skipFullSync            bool
+
+	useConnect bool
 }
 
 // WithRateLimiterConfig sets the RateLimiterConfig for a runner.
@@ -312,6 +314,14 @@ func WithRateLimiterConfig(cfg *ratelimitV1.RateLimiterConfig) Option {
 	}
 }
 
+// WithConnectExperiment uses the Connect transport for connector communication.
+func WithConnectExperiment() Option {
+	return func(ctx context.Context, w *runnerConfig) error {
+		w.useConnect = true
+		return nil
+	}
+}
+
 // WithExternalLimiter configures the connector to use an external rate limiter.
 // The `opts` map is injected into the environment in order for the service to be configured.
 func WithExternalLimiter(address string, opts map[string]string) Option {
@@ -556,9 +566,20 @@ func NewConnectorRunner(ctx context.Context, c types.ConnectorServer, opts ...Op
 		wrapperOpts = append(wrapperOpts, connector.WithFullSyncDisabled())
 	}
 
-	cw, err := connector.NewWrapper(ctx, c, wrapperOpts...)
-	if err != nil {
-		return nil, err
+	var cw types.ClientWrapper
+	var err error
+	if cfg.useConnect {
+		// FIXME(morgabra): Make this a top level option/select a port from the OS in the wrapper if unset.
+		wrapperOpts = append(wrapperOpts, connector.WithConnectListenPort(8000))
+		cw, err = connector.NewWrapperConnect(ctx, c, wrapperOpts...)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		cw, err = connector.NewWrapper(ctx, c, wrapperOpts...)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	runner.cw = cw

pkg/field/defaults.go

@@ -31,6 +31,7 @@ var (
 	ticketTemplatePathField    = StringField("ticket-template-path", WithHidden(true), WithDescription("A JSON file describing the ticket to create"), WithPersistent(true))
 	logLevelField              = StringField("log-level", WithDefaultValue("info"), WithDescription("The log level: debug, info, warn, error"), WithPersistent(true))
 	skipFullSync               = BoolField("skip-full-sync", WithDescription("This must be set to skip a full sync"), WithPersistent(true))
+	useConnectExperimental     = BoolField("use-connect-experiment", WithDescription("This must be set to enable the experimental connect transport"), WithHidden(true), WithPersistent(true))
 )
 
 // DefaultFields list the default fields expected in every single connector.
@@ -62,6 +63,7 @@ var DefaultFields = []SchemaField{
 	ticketTemplatePathField,
 	logLevelField,
 	skipFullSync,
+	useConnectExperimental,
 }
 
 func IsFieldAmongDefaultList(f SchemaField) bool {