Add IXP and country stats

Update version
Skip imports of any IXP with a non-ISO code

README.md

@@ -48,6 +48,10 @@ The backfill currently process a single month at a time and will look for the ea
 
 IMPORTANT NOTE: due to the way the code tries to figure out when a member left an IXP, you should run the backfill strictly in date order and *before* syncing the current data.
 
+## IXP stats
+
+The import process also generates monthly stats per IXP and per country. These are generated as of the 1st of the month used to import the data.
+
 ## Running programmatically
 
 If you'd like to run the import from code, rather than from the management command, you can call `importers.import_data()` directly.

ixp_tracker/importers.py

@@ -3,10 +3,11 @@
 from json.decoder import JSONDecodeError
 import logging
 from datetime import datetime, timedelta, timezone
-from typing import Callable, Protocol
+from typing import Callable, List, Protocol
 
 import requests
 import dateutil.parser
+from django_countries import countries
 
 from ixp_tracker.conf import IXP_TRACKER_PEERING_DB_KEY, IXP_TRACKER_PEERING_DB_URL, DATA_ARCHIVE_URL
 from ixp_tracker import models
@@ -22,6 +23,9 @@ def get_iso2_country(self, asn: int, as_at: datetime) -> str:
     def get_status(self, asn: int, as_at: datetime) -> str:
         pass
 
+    def get_asns_for_country(self, country: str, as_at: datetime) -> List[int]:
+        pass
+
 
 def import_data(
         geo_lookup: ASNGeoLookup,
@@ -99,6 +103,10 @@ def import_ixps(processing_date) -> bool:
 def process_ixp_data(processing_date: datetime):
     def do_process_ixp_data(all_ixp_data):
         for ixp_data in all_ixp_data:
+            country_data = countries.alpha2(ixp_data["country"])
+            if len(country_data) == 0:
+                logger.warning("Skipping IXP import as country code not found", extra={"country": ixp_data["country"], "id": ixp_data["id"]})
+                continue
             try:
                 models.IXP.objects.update_or_create(
                     peeringdb_id=ixp_data["id"],
@@ -108,7 +116,7 @@ def do_process_ixp_data(all_ixp_data):
                         "city": ixp_data["city"],
                         "website": ixp_data["website"],
                         "active_status": True,
-                        "country": ixp_data["country"],
+                        "country_code": ixp_data["country"],
                         "created": ixp_data["created"],
                         "last_updated": ixp_data["updated"],
                         "last_active": processing_date,
@@ -142,7 +150,7 @@ def process_asn_paged_data(all_asn_data):
                         "name": asn_data["name"],
                         "number": asn,
                         "network_type": asn_data["info_type"],
-                        "registration_country": geo_lookup.get_iso2_country(asn, last_updated),
+                        "registration_country_code": geo_lookup.get_iso2_country(asn, last_updated),
                         "created": asn_data["created"],
                         "last_updated": last_updated,
                     }
@@ -193,7 +201,7 @@ def do_process_member_data(all_member_data):
             member.date_left = end_of_month
             member.save()
             logger.debug("Member flagged as left due to inactivity", extra={"member": member.asn.number})
-        candidates = models.IXPMember.objects.filter(date_left=None, asn__registration_country="ZZ").all()
+        candidates = models.IXPMember.objects.filter(date_left=None, asn__registration_country_code="ZZ").all()
         for candidate in candidates:
             if geo_lookup.get_status(candidate.asn.number, processing_date) != "assigned":
                 end_of_last_month_active = (candidate.last_active.replace(day=1) - timedelta(days=1))

ixp_tracker/management/commands/ixp_tracker_import.py

@@ -2,11 +2,13 @@
 import logging
 import traceback
 from datetime import datetime, timezone
+from typing import List
 
 from django.core.management import BaseCommand
 
 from ixp_tracker.conf import IXP_TRACKER_GEO_LOOKUP_FACTORY
 from ixp_tracker.importers import ASNGeoLookup, import_data
+from ixp_tracker.stats import generate_stats
 
 logger = logging.getLogger("ixp_tracker")
 
@@ -19,6 +21,9 @@ def get_iso2_country(self, asn: int, as_at: datetime) -> str:
     def get_status(self, asn: int, as_at: datetime) -> str:
         return "assigned"
 
+    def get_asns_for_country(self, country: str, as_at: datetime) -> List[int]:
+        pass
+
 
 def load_geo_lookup(geo_lookup_name):
     if geo_lookup_name is not None:
@@ -46,6 +51,7 @@ def handle(self, *args, **options):
             geo_lookup = load_geo_lookup(IXP_TRACKER_GEO_LOOKUP_FACTORY) or DefaultASNGeoLookup()
             reset = options["reset_asns"]
             backfill_date = options["backfill"]
+            processing_date = None
             if backfill_date is None:
                 import_data(geo_lookup, reset)
             else:
@@ -54,6 +60,8 @@ def handle(self, *args, **options):
                     logger.warning("The --reset option has no effect when running a backfill")
                 import_data(geo_lookup, False, processing_date)
 
+            logger.debug("Generating stats")
+            generate_stats(geo_lookup, processing_date)
             logger.info("Import finished")
         except Exception as e:
             logging.error("Failed to import data", extra={"error": str(e), "trace": traceback.format_exc()})

ixp_tracker/migrations/0005_statspercountry_statsperixp_and_more.py

@@ -0,0 +1,55 @@
+# Generated by Django 5.0.8 on 2024-09-16 13:56
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('ixp_tracker', '0004_ixpmember_ixpmember_unique_ixp_membership'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='StatsPerCountry',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('country', models.CharField(max_length=2)),
+                ('stats_date', models.DateField()),
+                ('asn_count', models.IntegerField()),
+                ('member_count', models.IntegerField()),
+                ('asns_ixp_member_rate', models.FloatField()),
+                ('total_capacity', models.FloatField()),
+            ],
+            options={
+                'verbose_name': 'Per-country stats',
+            },
+        ),
+        migrations.CreateModel(
+            name='StatsPerIXP',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('stats_date', models.DateField()),
+                ('capacity', models.FloatField()),
+                ('members', models.IntegerField()),
+                ('local_asns_members_rate', models.FloatField()),
+            ],
+            options={
+                'verbose_name': 'IXP stats',
+            },
+        ),
+        migrations.AddConstraint(
+            model_name='statspercountry',
+            constraint=models.UniqueConstraint(fields=('country', 'stats_date'), name='ixp_tracker_unique_per_country_stats'),
+        ),
+        migrations.AddField(
+            model_name='statsperixp',
+            name='ixp',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='ixp_tracker.ixp'),
+        ),
+        migrations.AddConstraint(
+            model_name='statsperixp',
+            constraint=models.UniqueConstraint(fields=('ixp', 'stats_date'), name='ixp_tracker_unique_ixp_stats'),
+        ),
+    ]

ixp_tracker/migrations/0006_remove_statspercountry_ixp_tracker_unique_per_country_stats_and_more.py

@@ -0,0 +1,36 @@
+# Generated by Django 5.0.8 on 2024-09-17 13:23
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('ixp_tracker', '0005_statspercountry_statsperixp_and_more'),
+    ]
+
+    operations = [
+        migrations.RemoveConstraint(
+            model_name='statspercountry',
+            name='ixp_tracker_unique_per_country_stats',
+        ),
+        migrations.RenameField(
+            model_name='asn',
+            old_name='registration_country',
+            new_name='registration_country_code',
+        ),
+        migrations.RenameField(
+            model_name='ixp',
+            old_name='country',
+            new_name='country_code',
+        ),
+        migrations.RenameField(
+            model_name='statspercountry',
+            old_name='country',
+            new_name='country_code',
+        ),
+        migrations.AddConstraint(
+            model_name='statspercountry',
+            constraint=models.UniqueConstraint(fields=('country_code', 'stats_date'), name='ixp_tracker_unique_per_country_stats'),
+        ),
+    ]

ixp_tracker/models.py

@@ -9,7 +9,7 @@ class IXP(models.Model):
     website = models.URLField(null=True)
     active_status = models.BooleanField(default=True)
     peeringdb_id = models.IntegerField(null=True)
-    country = models.CharField(max_length=2)
+    country_code = models.CharField(max_length=2)
     created = models.DateTimeField()
     last_updated = models.DateTimeField()
     last_active = models.DateTimeField(null=True)
@@ -41,7 +41,7 @@ class ASN(models.Model):
     number = models.IntegerField()
     peeringdb_id = models.IntegerField(null=True)
     network_type = models.CharField(max_length=200, choices=NETWORK_TYPE_CHOICES, default='not-disclosed')
-    registration_country = models.CharField(max_length=2)
+    registration_country_code = models.CharField(max_length=2)
     created = models.DateTimeField()
     last_updated = models.DateTimeField()
 
@@ -68,7 +68,7 @@ class IXPMember(models.Model):
     last_active = models.DateTimeField(null=True)
 
     def __str__(self):
-        return self.ixp.name + " - " + self.asn.name
+        return f"{self.ixp.name} - {self.asn.name}"
 
     class Meta:
         verbose_name = "IXP Member"
@@ -77,3 +77,39 @@ class Meta:
         constraints = [
             models.UniqueConstraint(fields=['ixp', 'asn'], name='ixp_tracker_unique_ixp_membership')
         ]
+
+
+class StatsPerIXP(models.Model):
+    ixp = models.ForeignKey(IXP, on_delete=models.CASCADE)
+    stats_date = models.DateField()
+    capacity = models.FloatField()
+    members = models.IntegerField()
+    local_asns_members_rate = models.FloatField()
+
+    def __str__(self):
+        return f"{self.ixp.name} - {self.stats_date}"
+
+    class Meta:
+        verbose_name = "IXP stats"
+
+        constraints = [
+            models.UniqueConstraint(fields=['ixp', 'stats_date'], name='ixp_tracker_unique_ixp_stats')
+        ]
+
+class StatsPerCountry(models.Model):
+    country_code = models.CharField(max_length=2)
+    stats_date = models.DateField()
+    asn_count = models.IntegerField()
+    member_count = models.IntegerField()
+    asns_ixp_member_rate = models.FloatField()
+    total_capacity = models.FloatField()
+
+    def __str__(self):
+        return f"{self.country_code}-{self.stats_date}-{self.asn_count}-{self.member_count}"
+
+    class Meta:
+        verbose_name = "Per-country stats"
+
+        constraints = [
+            models.UniqueConstraint(fields=['country_code', 'stats_date'], name='ixp_tracker_unique_per_country_stats')
+        ]

ixp_tracker/stats.py

@@ -0,0 +1,91 @@
+import logging
+from datetime import datetime, timezone
+from typing import Dict, List, TypedDict, Union
+
+from django_countries import countries
+from django.db.models import Q
+
+from ixp_tracker.importers import ASNGeoLookup
+from ixp_tracker.models import IXP, IXPMember, StatsPerCountry, StatsPerIXP
+
+logger = logging.getLogger("ixp_tracker")
+
+
+class CountryStats(TypedDict):
+    all_asns: Union[List[int], None]
+    member_asns: set
+    total_capacity: int
+
+
+def generate_stats(geo_lookup: ASNGeoLookup, stats_date: datetime = None):
+    stats_date = stats_date or datetime.utcnow().replace(tzinfo=timezone.utc)
+    stats_date = stats_date.replace(day=1)
+    ixps = IXP.objects.filter(created__lte=stats_date).all()
+    all_members = (IXPMember.objects
+                   .filter(member_since__lte=stats_date)
+                   .filter(Q(date_left=None) | Q(date_left__gte=stats_date))).all()
+    all_stats_per_country: Dict[str, CountryStats] = {}
+    for code, _ in list(countries):
+        all_stats_per_country[code] = {
+            "all_asns": None,
+            "member_asns": set(),
+            "total_capacity": 0
+        }
+    for ixp in ixps:
+        logger.debug("Calculating growth stats for IXP", extra={"ixp": ixp.id})
+        members = [member for member in all_members if member.ixp == ixp]
+        member_count = len(members)
+        capacity = sum(member.speed for member in members)
+        ixp_country = ixp.country_code
+        country_stats = all_stats_per_country.get(ixp_country)
+        if country_stats is None:
+            logger.warning("Country not found", extra={"country": ixp_country})
+            country_stats = {
+                "all_asns": None,
+                "member_asns": set(),
+                "total_capacity": 0
+            }
+            all_stats_per_country[ixp_country] = country_stats
+        if country_stats.get("all_asns") is None:
+            all_stats_per_country[ixp_country]["all_asns"] = geo_lookup.get_asns_for_country(ixp_country, stats_date)
+        member_asns = [member.asn.number for member in members]
+        local_asns_members_rate = calculate_local_asns_members_rate(member_asns, all_stats_per_country[ixp_country]["all_asns"])
+        StatsPerIXP.objects.update_or_create(
+            ixp=ixp,
+            stats_date=stats_date.date(),
+            defaults={
+                "ixp": ixp,
+                "stats_date": stats_date.date(),
+                "members": member_count,
+                "capacity": (capacity/1000),
+                "local_asns_members_rate": local_asns_members_rate,
+            }
+        )
+        # We only count unique ASNs that are members of an IXP in a country
+        all_stats_per_country[ixp_country]["member_asns"] |= set(member_asns)
+        # But we count capacity for all members, i.e. an ASN member at 2 IXPs will have capacity at each included in the sum
+        all_stats_per_country[ixp_country]["total_capacity"] += capacity
+    for code, _ in list(countries):
+        country_stats = all_stats_per_country[code]
+        if country_stats.get("all_asns") is None:
+            country_stats["all_asns"] = geo_lookup.get_asns_for_country(code, stats_date)
+        local_asns_members_rate = calculate_local_asns_members_rate(country_stats["member_asns"], country_stats["all_asns"])
+        StatsPerCountry.objects.update_or_create(
+            country_code=code,
+            stats_date=stats_date.date(),
+            defaults={
+                "asn_count": len(country_stats["all_asns"]),
+                "member_count": len(country_stats["member_asns"]),
+                "asns_ixp_member_rate": local_asns_members_rate,
+                "total_capacity": (country_stats["total_capacity"]/1000),
+            }
+        )
+
+
+def calculate_local_asns_members_rate(member_asns: List[int], country_asns: [int]) -> float:
+    if len(country_asns) == 0:
+        return 0
+    # Ignore the current country for a member ASN (as that might have changed) but just get all current members
+    # that are in the list of ASNs registered to the country at the time
+    members_in_country = [asn for asn in member_asns if asn in country_asns]
+    return len(members_in_country) / len(country_asns)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "django-ixp-tracker"
-version = "0.2"
+version = "0.3"
 description = "Library to retrieve and manipulate data about IXPs"
 readme = "README.md"
 requires-python = ">=3.8"
@@ -14,6 +14,7 @@ classifiers = [
 ]
 dependencies = [
     "django>4.2",
+    "django-countries>7.6",
     "python-dateutil"
 ]
 

tests/test_asns_import.py

@@ -65,7 +65,7 @@ def test_updates_existing_data():
         number=dummy_asn_data["asn"],
         peeringdb_id=dummy_asn_data["id"],
         network_type="other",
-        registration_country="CH",
+        registration_country_code="CH",
         created="2019-01-01",
         last_updated="2024-05-01"
     )
@@ -85,7 +85,7 @@ def test_updates_existing_data():
         assert len(asns) == 1
         updated = asns.filter(peeringdb_id=dummy_asn_data["id"]).first()
         assert updated.name == "New ASN"
-        assert updated.registration_country == "AU"
+        assert updated.registration_country_code == "AU"
 
 
 def test_handles_errors_with_source_data():