forked from cockroachdb/cockroach
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
externalconn,changefeedccl: support
kafka in external connections
This change adds the ability to create an external connection that rerpresents a `kafka` sink. It also teaches changefeeds to recognize the `external` schema URI. When creating an external connection that represents a kafka sink, we run validation on the passed in URI before persisting it in the system table. This PR does not add any `WITH` options to the create statement but in the near future we will want to allow the user to pass in certain sink/resource specific configurations that will apply to all users of the external connection object. For example, a kafka JSON config. A changefeed can now be run to an `external` scheme URI that points to an existing external connection object. Since we can only represent kafka sinks as external connections as of now, the statement will only accept the options that are valid for a kafka sink. Informs: cockroachdb#84753 Release note (sql change): Users can now `CREATE EXTERNAL CONNECTION` to represent a `kafka` sink. Subsequently, users can run `CREATE CHANGEFEED` with an `external:///<external-connection-object-name` URI as the sink to use the kafka resource represented by the external connection object.
- Loading branch information
1 parent
12daec6
commit 3ed3fe0
Showing
18 changed files
with
490 additions
and
63 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| // Copyright 2022 The Cockroach Authors. | ||
| // | ||
| // Licensed as a CockroachDB Enterprise file under the Cockroach Community | ||
| // License (the "License"); you may not use this file except in compliance with | ||
| // the License. You may obtain a copy of the License at | ||
| // | ||
| // https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt | ||
|
|
||
| package changefeedccl | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/cloud/externalconn" | ||
| "github.com/cockroachdb/cockroach/pkg/cloud/externalconn/connectionpb" | ||
| "github.com/cockroachdb/cockroach/pkg/kv" | ||
| "github.com/cockroachdb/cockroach/pkg/security/username" | ||
| "github.com/cockroachdb/cockroach/pkg/sql/sqlutil" | ||
| "github.com/cockroachdb/errors" | ||
| ) | ||
|
|
||
| func makeExternalConnectionSink( | ||
| ctx context.Context, | ||
| u sinkURL, | ||
| user username.SQLUsername, | ||
| db *kv.DB, | ||
| ie sqlutil.InternalExecutor, | ||
| sinkCtx SinkContext, | ||
| ) (Sink, error) { | ||
| if u.Host == "" { | ||
| return nil, errors.Newf("host component of an external URI must refer to an "+ | ||
| "existing External Connection object: %s", u.String()) | ||
| } | ||
|
|
||
| externalConnectionName := u.Host | ||
|
|
||
| // TODO(adityamaru): Use the `user` in `cfg` to perform privilege checks on | ||
| // the external connection object we are about to retrieve. | ||
|
|
||
| // Retrieve the external connection object from the system table. | ||
| var ec *externalconn.ExternalConnection | ||
| if err := db.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error { | ||
| var err error | ||
| ec, err = externalconn.LoadExternalConnection(ctx, externalConnectionName, | ||
| connectionpb.TypeStorage, ie, user, txn) | ||
| return err | ||
| }); err != nil { | ||
| return nil, errors.Wrap(err, "failed to load external connection object") | ||
| } | ||
|
|
||
| // Construct a Sink handle for the underlying resource represented by the | ||
| // external connection object. | ||
| details := ec.ConnectionDetails() | ||
| connDetails, err := externalconn.MakeConnectionDetails(ctx, *details) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| connection, err := connDetails.Dial(ctx, sinkCtx, "" /* subdir */) | ||
| if err != nil { | ||
| return nil, errors.Wrap(err, "failed to Dial external connection") | ||
| } | ||
|
|
||
| var sink Sink | ||
| var ok bool | ||
| if sink, ok = connection.(Sink); !ok { | ||
| return nil, errors.AssertionFailedf("cannot convert Connection to Sink") | ||
| } | ||
|
|
||
| return sink, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| // Copyright 2022 The Cockroach Authors. | ||
| // | ||
| // Licensed as a CockroachDB Enterprise file under the Cockroach Community | ||
| // License (the "License"); you may not use this file except in compliance with | ||
| // the License. You may obtain a copy of the License at | ||
| // | ||
| // https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt | ||
|
|
||
| package changefeedccl | ||
|
|
||
| import ( | ||
| "context" | ||
| "net/url" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase" | ||
| "github.com/cockroachdb/cockroach/pkg/cloud/externalconn" | ||
| "github.com/cockroachdb/cockroach/pkg/cloud/externalconn/connectionpb" | ||
| "github.com/cockroachdb/errors" | ||
| ) | ||
|
|
||
| func parseAndValidateKafkaSinkURI( | ||
| ctx context.Context, uri *url.URL, | ||
| ) (connectionpb.ConnectionDetails, error) { | ||
| // Validate the kafka URI by creating a kafka sink. | ||
| // | ||
| // TODO(adityamaru): Add `CREATE EXTERNAL CONNECTION ... WITH` support to | ||
| // accept JSONConfig. | ||
| _, err := makeKafkaSink(ctx, sinkURL{URL: uri}, changefeedbase.Targets{}, "", nilMetricsRecorderBuilder) | ||
| if err != nil { | ||
| return connectionpb.ConnectionDetails{}, errors.Wrap(err, "invalid Kafka URI") | ||
| } | ||
|
|
||
| connDetails := connectionpb.ConnectionDetails{ | ||
| Provider: connectionpb.ConnectionProvider_kafka, | ||
| Details: &connectionpb.ConnectionDetails_Kafka{ | ||
| Kafka: &connectionpb.KafkaConnectionDetails{ | ||
| URI: uri.String(), | ||
| }, | ||
| }, | ||
| } | ||
| return connDetails, nil | ||
| } | ||
|
|
||
| type kafkaConnectionDetails struct { | ||
| connectionpb.ConnectionDetails | ||
| } | ||
|
|
||
| // Dial implements the ConnectionDetails interface. | ||
| func (k *kafkaConnectionDetails) Dial( | ||
| ctx context.Context, connectionCtx interface{}, _ string, | ||
| ) (externalconn.Connection, error) { | ||
| sinkCtx, ok := connectionCtx.(SinkContext) | ||
| if !ok { | ||
| return nil, errors.Newf("Kafka sink dialed with an incompatible context of type %T", connectionCtx) | ||
| } | ||
| uri, err := url.Parse(k.GetKafka().URI) | ||
| if err != nil { | ||
| return nil, errors.New("failed to parse kafka URI when dialing external connection") | ||
| } | ||
|
|
||
| // TODO(adityamaru): Currently, we're getting the kafkaJSONConfig from the | ||
| // `CREATE CHANGEFEED` statement but we might consider moving this option to a | ||
| // `CREATE EXTERNAL CONNECTION` option in the future. | ||
| return makeKafkaSink(ctx, sinkURL{URL: uri}, sinkCtx.targets, sinkCtx.kafkaJSONConfig, sinkCtx.mb) | ||
| } | ||
|
|
||
| // ConnectionProto implements the ConnectionDetails interface. | ||
| func (k *kafkaConnectionDetails) ConnectionProto() *connectionpb.ConnectionDetails { | ||
| return &k.ConnectionDetails | ||
| } | ||
|
|
||
| // ConnectionType implements the ConnectionDetails interface. | ||
| func (k *kafkaConnectionDetails) ConnectionType() connectionpb.ConnectionType { | ||
| return k.ConnectionDetails.Type() | ||
| } | ||
|
|
||
| var _ externalconn.ConnectionDetails = &kafkaConnectionDetails{} | ||
|
|
||
| func makeKafkaSinkConnectionDetails( | ||
| _ context.Context, details connectionpb.ConnectionDetails, | ||
| ) externalconn.ConnectionDetails { | ||
| return &kafkaConnectionDetails{ConnectionDetails: details} | ||
| } | ||
|
|
||
| func init() { | ||
| externalconn.RegisterConnectionDetailsFromURIFactory(connectionpb.ConnectionProvider_kafka, | ||
| changefeedbase.SinkSchemeKafka, parseAndValidateKafkaSinkURI, makeKafkaSinkConnectionDetails) | ||
| } |
Oops, something went wrong.