-
Notifications
You must be signed in to change notification settings - Fork 71
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
6 changed files
with
249 additions
and
17 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,111 @@ | ||
| package kafka | ||
|
|
||
| import "github.com/batchcorp/plumber/cli" | ||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/jhump/protoreflect/desc" | ||
| "github.com/pkg/errors" | ||
| "github.com/relistan/go-director" | ||
| "github.com/sirupsen/logrus" | ||
|
|
||
| "github.com/batchcorp/plumber/api" | ||
| "github.com/batchcorp/plumber/backends/kafka/types" | ||
| "github.com/batchcorp/plumber/cli" | ||
| "github.com/batchcorp/plumber/relay" | ||
| ) | ||
|
|
||
| type Relayer struct { | ||
| Options *cli.Options | ||
| MsgDesc *desc.MessageDescriptor | ||
| RelayCh chan interface{} | ||
| log *logrus.Entry | ||
| Looper *director.FreeLooper | ||
| DefaultContext context.Context | ||
| } | ||
|
|
||
| type IKafkaRelayer interface { | ||
| Relay() error | ||
| } | ||
|
|
||
| var ( | ||
| errMissingTopic = errors.New("You must specify a topic") | ||
| ) | ||
|
|
||
| // Relay sets up a new Kafka relayer, starts GRPC workers and the API server | ||
| func Relay(opts *cli.Options) error { | ||
| panic("not implemented") | ||
| if err := validateRelayOptions(opts); err != nil { | ||
| return errors.Wrap(err, "unable to verify options") | ||
| } | ||
|
|
||
| // Create new relayer instance (+ validate token & gRPC address) | ||
| relayCfg := &relay.Config{ | ||
| Token: opts.RelayToken, | ||
| GRPCAddress: opts.RelayGRPCAddress, | ||
| NumWorkers: opts.RelayNumWorkers, | ||
| Timeout: opts.RelayGRPCTimeout, | ||
| RelayCh: make(chan interface{}, 1), | ||
| DisableTLS: opts.RelayGRPCDisableTLS, | ||
| } | ||
|
|
||
| grpcRelayer, err := relay.New(relayCfg) | ||
| if err != nil { | ||
| return errors.Wrap(err, "unable to create new gRPC relayer") | ||
| } | ||
|
|
||
| // Launch HTTP server | ||
| go func() { | ||
| if err := api.Start(opts.RelayHTTPListenAddress, opts.Version); err != nil { | ||
| logrus.Fatalf("unable to start API server: %s", err) | ||
| } | ||
| }() | ||
|
|
||
| if err := grpcRelayer.StartWorkers(); err != nil { | ||
| return errors.Wrap(err, "unable to start gRPC relay workers") | ||
| } | ||
|
|
||
| r := &Relayer{ | ||
| Options: opts, | ||
| RelayCh: relayCfg.RelayCh, | ||
| log: logrus.WithField("pkg", "kafka/relay"), | ||
| Looper: director.NewFreeLooper(director.FOREVER, make(chan error)), | ||
| DefaultContext: context.Background(), | ||
| } | ||
|
|
||
| return r.Relay() | ||
| } | ||
|
|
||
| // validateRelayOptions ensures all required CLI options are present before initializing relay mode | ||
| func validateRelayOptions(opts *cli.Options) error { | ||
| if opts.Kafka.Topic == "" { | ||
| return errMissingTopic | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Relay reads messages from Kafka and sends them to RelayCh which is then read by relay.Run() | ||
| func (r *Relayer) Relay() error { | ||
| r.log.Infof("Relaying Kafka messages from '%s' topic -> '%s'", | ||
| r.Options.Kafka.Topic, r.Options.RelayGRPCAddress) | ||
|
|
||
| r.log.Infof("HTTP server listening on '%s'", r.Options.RelayHTTPListenAddress) | ||
|
|
||
| reader, err := NewReader(r.Options) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| defer reader.Reader.Close() | ||
| defer reader.Conn.Close() | ||
|
|
||
| for { | ||
| msg, err := reader.Reader.ReadMessage(r.DefaultContext) | ||
| if err != nil { | ||
| r.log.Errorf("Unable to read message: %s", err) | ||
| continue | ||
| } | ||
| r.log.Infof("Writing Kafka message to relay channel: %s", msg.Value) | ||
| r.RelayCh <- &types.RelayMessage{ | ||
| Value: &msg, | ||
| Options: &types.RelayMessageOptions{}, | ||
| } | ||
| } | ||
| } |
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,15 @@ | ||
| package types | ||
|
|
||
| import ( | ||
| "github.com/segmentio/kafka-go" | ||
| ) | ||
|
|
||
| // RelayMessage encapsulates a kafka message that is read by relay.Run() | ||
| type RelayMessage struct { | ||
| Value *kafka.Message | ||
| Options *RelayMessageOptions | ||
| } | ||
|
|
||
| // RelayMessageOptions contains any additional options necessary for processing of Kafka messages by the relayer | ||
| type RelayMessageOptions struct { | ||
| } |
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,65 @@ | ||
| package relay | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/batchcorp/schemas/build/go/events/records" | ||
| "github.com/batchcorp/schemas/build/go/services" | ||
| "github.com/pkg/errors" | ||
| "github.com/segmentio/kafka-go" | ||
| "google.golang.org/grpc" | ||
|
|
||
| "github.com/batchcorp/plumber/backends/kafka/types" | ||
| ) | ||
|
|
||
| var ( | ||
| errMissingMessage = errors.New("msg cannot be nil") | ||
| errMissingMessageValue = errors.New("msg.Value cannot be nil") | ||
| ) | ||
|
|
||
| // handleKafka sends a Kafka relay message to the GRPC server | ||
| func (r *Relay) handleKafka(ctx context.Context, conn *grpc.ClientConn, msg *types.RelayMessage) error { | ||
| if err := r.validateKafkaRelayMessage(msg); err != nil { | ||
| return errors.Wrap(err, "unable to validate kafka relay message") | ||
| } | ||
|
|
||
| kafkaRecord := convertKafkaMessageToProtobufRecord(msg.Value) | ||
|
|
||
| client := services.NewGRPCCollectorClient(conn) | ||
|
|
||
| if _, err := client.AddKafkaRecord(ctx, &services.KafkaSinkRecordRequest{ | ||
| Records: []*records.KafkaSinkRecord{kafkaRecord}, | ||
| }); err != nil { | ||
| r.log.Errorf("%+v", kafkaRecord) | ||
| return errors.Wrap(err, "unable to complete AddKafkaRecord call") | ||
| } | ||
| r.log.Debug("successfully handled kafka message") | ||
| return nil | ||
| } | ||
|
|
||
| // validateKafkaRelayMessage ensures all necessary values are present for a Kafka relay message | ||
| func (r *Relay) validateKafkaRelayMessage(msg *types.RelayMessage) error { | ||
| if msg == nil { | ||
| return errMissingMessage | ||
| } | ||
|
|
||
| if msg.Value == nil { | ||
| return errMissingMessageValue | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // convertKafkaMessageToProtobufRecord creates a records.KafkaSinkRecord from a kafka.Message which can then | ||
| // be sent to the GRPC server | ||
| func convertKafkaMessageToProtobufRecord(msg *kafka.Message) *records.KafkaSinkRecord { | ||
| return &records.KafkaSinkRecord{ | ||
| Topic: msg.Topic, | ||
| Key: msg.Key, | ||
| Value: msg.Value, | ||
| Timestamp: time.Now().UTC().UnixNano(), | ||
| Offset: msg.Offset, | ||
| Partition: int32(msg.Partition), | ||
| } | ||
| } |
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