prysm-pulse/beacon-chain/db/kafka/export_wrapper.go

108 lines
3.0 KiB
Go
Raw Normal View History

// Package kafka defines an implementation of Database interface
// which exports streaming data using Kafka for data analysis.
package kafka
import (
"context"
fssz "github.com/ferranbt/fastssz"
"github.com/prysmaticlabs/prysm/beacon-chain/db/iface"
"github.com/prysmaticlabs/prysm/proto/interfaces"
"github.com/prysmaticlabs/prysm/shared/featureconfig"
"github.com/prysmaticlabs/prysm/shared/traceutil"
"go.opencensus.io/trace"
jsonpb "google.golang.org/protobuf/encoding/protojson"
"gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
Refactor dependencies, make Prysm "go gettable" (#6053) * Fix a few deps to work with go.mod, check in generated files * Update Gossipsub to 1.1 (#5998) * update libs * add new validators * add new deps * new set of deps * tls * further fix gossip update * get everything to build * clean up * gaz * fix build * fix all tests * add deps to images * imports Co-authored-by: rauljordan <raul@prysmaticlabs.com> * Beacon chain builds with go build * fix bazel * fix dep * lint * Add github action for testing go * on PR for any branch * fix libp2p test failure * Fix TestProcessBlock_PassesProcessingConditions by updating the proposer index in test * Revert "Fix TestProcessBlock_PassesProcessingConditions by updating the proposer index in test" This reverts commit 43676894ab01f03fe90a9b8ee3ecfbc2ec1ec4e4. * Compute and set proposer index instead of hard code * Add back go mod/sum, fix deps * go build ./... * Temporarily skip two tests * Fix kafka confluent patch * Fix kafka confluent patch * fix kafka build * fix kafka * Add info in DEPENDENCIES. Added a stub link for Why Bazel? until https://github.com/prysmaticlabs/documentation/issues/138 * Update fuzz ssz files as well * Update fuzz ssz files as well * getting closer * rollback rules_go and gazelle * fix gogo protobuf * install librdkafka-dev as part of github actions * Update kafka to a recent version where librkafkfa is not required for go modules * clarify comment * fix kafka build * disable go tests * comment * Fix geth dependencies for end to end * rename word * lint * fix docker Co-authored-by: Nishant Das <nishdas93@gmail.com> Co-authored-by: rauljordan <raul@prysmaticlabs.com> Co-authored-by: terence tsao <terence@prysmaticlabs.com>
2020-05-31 06:44:34 +00:00
_ "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka/librdkafka" // Required for c++ kafka library.
"gopkg.in/errgo.v2/fmt/errors"
)
var _ iface.Database = (*Exporter)(nil)
var marshaler = jsonpb.MarshalOptions{}
// Exporter wraps a database interface and exports certain objects to kafka topics.
type Exporter struct {
db iface.Database
p *kafka.Producer
}
// Wrap the db with kafka exporter. If the feature flag is not enabled, this service does not wrap
// the database, but returns the underlying database pointer itself.
func Wrap(db iface.Database) (iface.Database, error) {
if featureconfig.Get().KafkaBootstrapServers == "" {
log.Debug("Empty Kafka bootstrap servers list, database was not wrapped with Kafka exporter")
return db, nil
}
p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": featureconfig.Get().KafkaBootstrapServers})
if err != nil {
return nil, err
}
return &Exporter{db: db, p: p}, nil
}
func (e Exporter) publish(ctx context.Context, topic string, msg interfaces.SignedBeaconBlock) error {
ctx, span := trace.StartSpan(ctx, "kafka.publish")
defer span.End()
var err error
var buf []byte
if buf, err = marshaler.Marshal(msg.Proto()); err != nil {
traceutil.AnnotateError(span, err)
return err
}
var key [32]byte
if v, ok := msg.(fssz.HashRoot); ok {
key, err = v.HashTreeRoot()
} else {
err = errors.New("object does not follow hash tree root interface")
}
if err != nil {
traceutil.AnnotateError(span, err)
return err
}
if err := e.p.Produce(&kafka.Message{
TopicPartition: kafka.TopicPartition{
Topic: &topic,
},
Value: buf,
Key: key[:],
}, nil); err != nil {
traceutil.AnnotateError(span, err)
return err
}
return nil
}
// Close closes kafka producer and underlying db.
func (e Exporter) Close() error {
e.p.Close()
return e.db.Close()
}
// SaveBlock publishes to the kafka topic for beacon blocks.
func (e Exporter) SaveBlock(ctx context.Context, block interfaces.SignedBeaconBlock) error {
go func() {
if err := e.publish(ctx, "beacon_block", block); err != nil {
log.WithError(err).Error("Failed to publish block")
}
}()
return e.db.SaveBlock(ctx, block)
}
// SaveBlocks publishes to the kafka topic for beacon blocks.
func (e Exporter) SaveBlocks(ctx context.Context, blocks []interfaces.SignedBeaconBlock) error {
go func() {
for _, block := range blocks {
if err := e.publish(ctx, "beacon_block", block); err != nil {
log.WithError(err).Error("Failed to publish block")
}
}
}()
return e.db.SaveBlocks(ctx, blocks)
}