feat(server/coordinator): add geoip data for node records

migrations/postgres/004_geo.down.sql

@@ -0,0 +1,9 @@
+ALTER TABLE node_record
+DROP COLUMN geo_city,
+DROP COLUMN geo_country,
+DROP COLUMN geo_country_code,
+DROP COLUMN geo_continent_code,
+DROP COLUMN geo_longitude,
+DROP COLUMN geo_latitude,
+DROP COLUMN geo_autonomous_system_number,
+DROP COLUMN geo_autonomous_system_organization;

migrations/postgres/004_geo.up.sql

@@ -0,0 +1,9 @@
+ALTER TABLE node_record
+ADD COLUMN geo_city VARCHAR(128),
+ADD COLUMN geo_country VARCHAR(128),
+ADD COLUMN geo_country_code VARCHAR(128),
+ADD COLUMN geo_continent_code VARCHAR(128),
+ADD COLUMN geo_longitude FLOAT,
+ADD COLUMN geo_latitude FLOAT,
+ADD COLUMN geo_autonomous_system_number INT,
+ADD COLUMN geo_autonomous_system_organization VARCHAR(128);

pkg/server/persistence/node/record.go

@@ -28,6 +28,22 @@ type Record struct {
  IP4 *string `json:"ip4" db:"ip4" fieldopt:"omitempty"`
  // IP6 is the IPv6 address of the node record.
  IP6 *string `json:"ip6" db:"ip6" fieldopt:"omitempty"`
+ // GeoCity is the city of the node record.
+ GeoCity *string `json:"geoCity" db:"geo_city" fieldopt:"omitempty"`
+ // GeoCountry is the country of the node record.
+ GeoCountry *string `json:"geoCountry" db:"geo_country" fieldopt:"omitempty"`
+ // GeoCountryCode is the country code of the node record.
+ GeoCountryCode *string `json:"geoCountryCode" db:"geo_country_code" fieldopt:"omitempty"`
+ // GeoContinentCode is the continent code of the node record.
+ GeoContinentCode *string `json:"geoContinentCode" db:"geo_continent_code" fieldopt:"omitempty"`
+ // GeoLongitude is the longitude of the node record.
+ GeoLongitude *float64 `json:"geoLongitude" db:"geo_longitude" fieldopt:"omitempty"`
+ // GeoLatitude is the latitude of the node record.
+ GeoLatitude *float64 `json:"geoLatitude" db:"geo_latitude" fieldopt:"omitempty"`
+ // GeoAutonomousSystemNumber is the autonomous system number of the node record.
+ GeoAutonomousSystemNumber *uint32 `json:"geoAutonomousSystemNumber" db:"geo_autonomous_system_number" fieldopt:"omitempty"`
+ // GeoAutonomousSystemOrganization is the autonomous system organization of the node record.
+ GeoAutonomousSystemOrganization *string `json:"geoAutonomousSystemOrganization" db:"geo_autonomous_system_organization" fieldopt:"omitempty"`
  // TCP4 is the TCP port of the node record.
  TCP4 *uint32 `json:"tcp4" db:"tcp4" fieldopt:"omitempty"`
  // TCP6 is the TCP port of the node record.

pkg/server/service/coordinator/client.go

@@ -5,10 +5,12 @@ import (
  "database/sql"
  "fmt"
  "math/rand"
+ "net"
  "strings"
  "time"
 
  "github.com/ethpandaops/xatu/pkg/proto/xatu"
+ "github.com/ethpandaops/xatu/pkg/server/geoip"
  "github.com/ethpandaops/xatu/pkg/server/persistence"
  "github.com/ethpandaops/xatu/pkg/server/persistence/node"
  n "github.com/ethpandaops/xatu/pkg/server/service/coordinator/node"
@@ -23,16 +25,17 @@ const (
 type Client struct {
  xatu.UnimplementedCoordinatorServer
 
- log logrus.FieldLogger
- config *Config
- persistence *persistence.Client
+ log logrus.FieldLogger
+ config *Config
+ persistence *persistence.Client
+ geoipProvider geoip.Provider
 
  metrics *Metrics
 
  nodeRecord *n.Record
 }
 
-func NewClient(ctx context.Context, log logrus.FieldLogger, conf *Config, p *persistence.Client) (*Client, error) {
+func NewClient(ctx context.Context, log logrus.FieldLogger, conf *Config, p *persistence.Client, geoipProvider geoip.Provider) (*Client, error) {
  if p == nil {
  return nil, fmt.Errorf("%s: persistence is required", ServiceType)
  }
@@ -45,11 +48,12 @@ func NewClient(ctx context.Context, log logrus.FieldLogger, conf *Config, p *per
  }
 
  e := &Client{
- log: logger,
- config: conf,
- persistence: p,
- nodeRecord: nodeRecord,
- metrics: NewMetrics("xatu_server_coordinator"),
+ log: logger,
+ config: conf,
+ persistence: p,
+ geoipProvider: geoipProvider,
+ nodeRecord: nodeRecord,
+ metrics: NewMetrics("xatu_server_coordinator"),
  }
 
  return e, nil
@@ -89,6 +93,34 @@ func (c *Client) CreateNodeRecords(ctx context.Context, req *xatu.CreateNodeReco
  return nil, err
  }
 
+ if c.geoipProvider != nil {
+ ipAddress := pRecord.IP4
+ if ipAddress == nil {
+ ipAddress = pRecord.IP6
+ }
+
+ if ipAddress != nil {
+ ip := net.ParseIP(*ipAddress)
+ if ip != nil {
+ geoipLookupResult, err := c.geoipProvider.LookupIP(ctx, ip)
+ if err != nil {
+ c.log.WithField("ip", *ipAddress).WithError(err).Warn("failed to lookup geoip data")
+ }
+
+ if geoipLookupResult != nil {
+ pRecord.GeoCity = &geoipLookupResult.CityName
+ pRecord.GeoCountry = &geoipLookupResult.CountryName
+ pRecord.GeoCountryCode = &geoipLookupResult.CountryCode
+ pRecord.GeoContinentCode = &geoipLookupResult.ContinentCode
+ pRecord.GeoLongitude = &geoipLookupResult.Longitude
+ pRecord.GeoLatitude = &geoipLookupResult.Latitude
+ pRecord.GeoAutonomousSystemNumber = &geoipLookupResult.AutonomousSystemNumber
+ pRecord.GeoAutonomousSystemOrganization = &geoipLookupResult.AutonomousSystemOrganization
+ }
+ }
+ }
+ }
+
  c.nodeRecord.Write(pRecord)
  }
 

pkg/server/service/service.go

@@ -50,7 +50,7 @@ func CreateGRPCServices(ctx context.Context, log logrus.FieldLogger, cfg *Config
  return nil, err
  }
 
- service, err := coordinator.NewClient(ctx, log, &cfg.Coordinator, p)
+ service, err := coordinator.NewClient(ctx, log, &cfg.Coordinator, p, g)
  if err != nil {
  return nil, err
  }