-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(cannon): Connect to Coordinator (#179)
* feat(coordinator): Add new Cannon types * feat: Split out derivers and hook up coordinator rpc * feat: Split out derivers and hook up coordinator rpc * refactor: set maximum interval for backoff to 1 minute
- Loading branch information
Showing
32 changed files
with
3,074 additions
and
1,530 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| package coordinator | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "net" | ||
|
|
||
| "github.com/ethpandaops/xatu/pkg/proto/xatu" | ||
| "github.com/sirupsen/logrus" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/credentials" | ||
| "google.golang.org/grpc/credentials/insecure" | ||
| "google.golang.org/grpc/encoding/gzip" | ||
| "google.golang.org/grpc/metadata" | ||
| ) | ||
|
|
||
| type Client struct { | ||
| config *Config | ||
| log logrus.FieldLogger | ||
|
|
||
| conn *grpc.ClientConn | ||
| pb xatu.CoordinatorClient | ||
| } | ||
|
|
||
| func New(config *Config, log logrus.FieldLogger) (*Client, error) { | ||
| if config == nil { | ||
| return nil, errors.New("config is required") | ||
| } | ||
|
|
||
| if err := config.Validate(); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| var opts []grpc.DialOption | ||
|
|
||
| if config.TLS { | ||
| host, _, err := net.SplitHostPort(config.Address) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("fail to get host from address: %v", err) | ||
| } | ||
|
|
||
| opts = append(opts, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, host))) | ||
| } else { | ||
| opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) | ||
| } | ||
|
|
||
| conn, err := grpc.Dial(config.Address, opts...) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("fail to dial: %v", err) | ||
| } | ||
|
|
||
| pbClient := xatu.NewCoordinatorClient(conn) | ||
|
|
||
| return &Client{ | ||
| config: config, | ||
| log: log, | ||
| conn: conn, | ||
| pb: pbClient, | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *Client) Start(ctx context.Context) error { | ||
| return nil | ||
| } | ||
|
|
||
| func (c *Client) Stop(ctx context.Context) error { | ||
| if err := c.conn.Close(); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (c *Client) GetCannonLocation(ctx context.Context, typ xatu.CannonType, networkID string) (*xatu.CannonLocation, error) { | ||
| req := xatu.GetCannonLocationRequest{ | ||
| Type: typ, | ||
| NetworkId: networkID, | ||
| } | ||
|
|
||
| md := metadata.New(c.config.Headers) | ||
| ctx = metadata.NewOutgoingContext(ctx, md) | ||
|
|
||
| res, err := c.pb.GetCannonLocation(ctx, &req, grpc.UseCompressor(gzip.Name)) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return res.Location, nil | ||
| } | ||
|
|
||
| func (c *Client) UpsertCannonLocationRequest(ctx context.Context, location *xatu.CannonLocation) error { | ||
| req := xatu.UpsertCannonLocationRequest{ | ||
| Location: location, | ||
| } | ||
|
|
||
| md := metadata.New(c.config.Headers) | ||
| ctx = metadata.NewOutgoingContext(ctx, md) | ||
|
|
||
| _, err := c.pb.UpsertCannonLocation(ctx, &req, grpc.UseCompressor(gzip.Name)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } |
Oops, something went wrong.