feat!(ingestors): filter assets per site

internal/config/settings.go

@@ -8,10 +8,16 @@ import (
 	"github.com/spf13/viper"
 )
 
+type Filter string
+
 const (
 	defaultListenAddress = "0.0.0.0"
 	defaultListenPort    = 8080
 	localPath            = "."
+
+	SiteFilter       Filter = "site"
+	SiteGroupFilter  Filter = "site_group"
+	SiteRegionFilter Filter = "region"
 )
 
 var (
@@ -21,8 +27,9 @@ var (
 type Config struct {
 	Authentication AuthConfig
 	NetBox         struct {
-		URL    string
-		APIKey string
+		URL                 string
+		APIKey              string
+		DatacenterFilterKey Filter
 	}
 	Log struct {
 		Level  string
@@ -61,6 +68,7 @@ func setDefaults() {
 
 	viper.SetDefault("NetBox.URL", "")
 	viper.SetDefault("NetBox.APIKey", "")
+	viper.SetDefault("NetBox.DatacenterFilterKey", SiteFilter)
 
 	viper.SetDefault("Build.Interval", time.Minute)
 	viper.SetDefault("Build.AllDevicesMustBuild", false)

internal/ingestor/cmdb/bgp_global.go

@@ -12,8 +12,9 @@ import (
 // GetBGPGlobal returns all BGP global configuration from the Network CMDB.
 func GetBGPGlobal() ([]*bgp.BGPGlobal, error) {
 	response := netbox.NetboxResponse[bgp.BGPGlobal]{}
+	params := deviceDatacenterFilter()
 
-	err := netbox.Get("/api/plugins/cmdb/bgp-global/", &response)
+	err := netbox.Get("/api/plugins/cmdb/bgp-global/", &response, params)
 	if err != nil {
 		return nil, fmt.Errorf("BGP Global fetching failure: %w", err)
 	}

internal/ingestor/cmdb/bgp_session.go

@@ -12,8 +12,9 @@ import (
 // GetBGPSessions returns all BGP sessions from the Network CMDB.
 func GetBGPSessions() ([]*bgp.Session, error) {
 	response := netbox.NetboxResponse[bgp.Session]{}
+	params := deviceDatacenterFilter()
 
-	err := netbox.Get("/api/plugins/cmdb/bgp-sessions/", &response)
+	err := netbox.Get("/api/plugins/cmdb/bgp-sessions/", &response, params)
 	if err != nil {
 		return nil, fmt.Errorf("BGP Sessions fetching failure: %w", err)
 	}

internal/ingestor/cmdb/community_list.go

@@ -12,8 +12,9 @@ import (
 // GetCommunityLists returns all community-lists from the Network CMDB.
 func GetCommunityLists() ([]*routingpolicy.CommunityList, error) {
 	response := netbox.NetboxResponse[routingpolicy.CommunityList]{}
+	params := deviceDatacenterFilter()
 
-	err := netbox.Get("/api/plugins/cmdb/bgp-community-lists/", &response)
+	err := netbox.Get("/api/plugins/cmdb/bgp-community-lists/", &response, params)
 	if err != nil {
 		return nil, fmt.Errorf("BGP Community Lists fetching failure: %w", err)
 	}

internal/ingestor/cmdb/peer_group.go

@@ -15,8 +15,9 @@ import (
 // You should migrate to configuration without using peer-groups.
 func GetPeerGroups() ([]*bgp.PeerGroup, error) {
 	response := netbox.NetboxResponse[bgp.PeerGroup]{}
+	params := deviceDatacenterFilter()
 
-	err := netbox.Get("/api/plugins/cmdb/peer-groups/", &response)
+	err := netbox.Get("/api/plugins/cmdb/peer-groups/", &response, params)
 	if err != nil {
 		return nil, fmt.Errorf("peer-groups fetching failure: %w", err)
 	}

internal/ingestor/cmdb/prefix_list.go

@@ -12,8 +12,9 @@ import (
 // GetPrefixLists returns all prefix-lists from the Network CMDB.
 func GetPrefixLists() ([]*routingpolicy.PrefixList, error) {
 	response := netbox.NetboxResponse[routingpolicy.PrefixList]{}
+	params := deviceDatacenterFilter()
 
-	err := netbox.Get("/api/plugins/cmdb/prefix-lists/", &response)
+	err := netbox.Get("/api/plugins/cmdb/prefix-lists/", &response, params)
 	if err != nil {
 		return nil, fmt.Errorf("prefix-lists fetching failure: %w", err)
 	}

internal/ingestor/cmdb/query_param.go

@@ -0,0 +1,28 @@
+package cmdb
+
+import (
+	"net/url"
+
+	"github.com/criteo/data-aggregation-api/internal/config"
+	"github.com/rs/zerolog/log"
+)
+
+func deviceDatacenterFilter() url.Values {
+	datacenterFilter := ""
+
+	switch string(config.Cfg.NetBox.DatacenterFilterKey) {
+	case "site":
+		datacenterFilter = "device__site__name"
+	case "site_group":
+		datacenterFilter = "device__site__group__name"
+	case "region":
+		datacenterFilter = "device__site__region__name"
+	default:
+		log.Fatal().Msgf("unknown datacenter filter: %s", config.Cfg.NetBox.DatacenterFilterKey)
+	}
+
+	params := url.Values{}
+	params.Set(datacenterFilter, config.Cfg.Datacenter)
+
+	return params
+}

internal/ingestor/cmdb/route_policy.go

@@ -12,8 +12,9 @@ import (
 // GetRoutePolicies returns all route-policies defined in the CDMB.
 func GetRoutePolicies() ([]*routingpolicy.RoutePolicy, error) {
 	response := netbox.NetboxResponse[routingpolicy.RoutePolicy]{}
+	params := deviceDatacenterFilter()
 
-	err := netbox.Get("/api/plugins/cmdb/route-policies/", &response)
+	err := netbox.Get("/api/plugins/cmdb/route-policies/", &response, params)
 	if err != nil {
 		return nil, fmt.Errorf("route-policies fetching failure: %w", err)
 	}

internal/ingestor/dcim/network_inventory.go

@@ -2,9 +2,11 @@ package dcim
 
 import (
 	"fmt"
+	"net/url"
 
 	"github.com/rs/zerolog/log"
 
+	"github.com/criteo/data-aggregation-api/internal/config"
 	"github.com/criteo/data-aggregation-api/internal/ingestor/netbox"
 	"github.com/criteo/data-aggregation-api/internal/model/dcim"
 )
@@ -15,8 +17,12 @@ import (
 func GetNetworkInventory() ([]*dcim.NetworkDevice, error) {
 	response := netbox.NetboxResponse[dcim.NetworkDevice]{}
 
-	if err := netbox.Get("/api/dcim/devices/?role__n=server", &response); err != nil {
-		return nil, fmt.Errorf("network inventory fetching failure: %s", err)
+	params := url.Values{}
+	params.Set(string(config.Cfg.NetBox.DatacenterFilterKey), config.Cfg.Datacenter)
+	params.Set("role__n", "server")
+
+	if err := netbox.Get("/api/dcim/devices/", &response, params); err != nil {
+		return nil, fmt.Errorf("network inventory fetching failure: %w", err)
 	}
 
 	if response.Count != len(response.Results) {

internal/ingestor/netbox/netbox.go

@@ -4,7 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"strings"
+	"net/url"
 	"time"
 
 	"github.com/go-playground/validator/v10"
@@ -31,35 +31,30 @@ func NewGetRequest(url string) (*http.Request, error) {
 }
 
 // Get fetches a Netbox endpoint.
-func Get[R any](endpoint string, out *NetboxResponse[R]) error {
+func Get[R any](endpoint string, out *NetboxResponse[R], params url.Values) error {
 	const endpointKey = "endpoint"
 	client := http.Client{Timeout: 10 * 60 * time.Second}
-	sep := "?"
 
-	// TODO: use url.Values and url.JoinPath instead
-	if strings.Contains(endpoint, sep) {
-		sep = "&"
-	}
+	params.Set("limit", "0")
+	params.Set("ordering", "id")
 
-	datacenterFilter := ""
-	// TODO: implement filter on CMDB side!
-	if config.Cfg.Datacenter != "" {
-		datacenterFilter = "&site_group=" + strings.ToUpper(config.Cfg.Datacenter)
+	baseURL, err := url.JoinPath(config.Cfg.NetBox.URL, endpoint)
+	if err != nil {
+		return fmt.Errorf("failed to assemble URL: %w", err)
 	}
-
-	url := config.Cfg.NetBox.URL + endpoint + sep + "limit=0&ordering=id" + datacenterFilter
+	url := baseURL + "?" + params.Encode()
 	log.Info().Str(endpointKey, endpoint).Msgf("Get %s", url)
 
 	// Get all pages
 	for url != "" {
 		req, err := NewGetRequest(url)
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to create request: %w", err)
 		}
 
 		data, err := client.Do(req)
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to query netbox: %w", err)
 		}
 		defer func() {
 			if err := data.Body.Close(); err != nil {
@@ -74,7 +69,7 @@ func Get[R any](endpoint string, out *NetboxResponse[R]) error {
 		var buffer NetboxResponse[R]
 		err = json.NewDecoder(data.Body).Decode(&buffer)
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to decode netbox response: %w", err)
 		}
 
 		out.Results = append(out.Results, buffer.Results...)

settings.example.yaml

@@ -20,6 +20,7 @@ Authentication:
 NetBox:
   URL: "https://netbox.local"
   APIKey: "<some_key>"
+  DatacenterFilterKey: "site_group"
 
 Build:
   Interval: "30m"