Add basic statistcs command.

Author: jcjones

partitionmanager/cli.py

@@ -22,6 +22,7 @@
  toSqlUrl,
  TableInformationException,
 )
+from partitionmanager.stats import get_statistics
 from partitionmanager.sql import SubprocessDatabaseCommand, IntegratedDatabaseCommand
 
 parser = argparse.ArgumentParser(
@@ -63,11 +64,12 @@ def from_argparse(self, args):
  self.dbcmd = IntegratedDatabaseCommand(args.dburl)
  else:
  self.dbcmd = SubprocessDatabaseCommand(args.mariadb)
- if args.days:
+ if "days" in args and args.days:
  self.partition_duration = timedelta(days=args.days)
  if self.partition_duration <= timedelta():
  raise ValueError("Negative lifespan is not allowed")
- self.noop = args.noop
+ if "noop" in args:
+ self.noop = args.noop
 
  def from_yaml_file(self, file):
  data = yaml.safe_load(file)
@@ -111,7 +113,7 @@ def partition_cmd(args):
  assert_table_is_compatible(conf.dbcmd, table)
  except TableInformationException as tie:
  logging.error(f"Cannot proceed: {tie}")
- return {}
+ return dict()
 
  all_results = dict()
  for table in conf.tables:
@@ -174,6 +176,38 @@ def partition_cmd(args):
 partition_parser.set_defaults(func=partition_cmd)
 
 
+def stats_cmd(args):
+ conf = Config()
+ conf.from_argparse(args)
+ if args.config:
+ conf.from_yaml_file(args.config)
+
+ # Preflight
+ try:
+ for table in conf.tables:
+ assert_table_is_compatible(conf.dbcmd, table)
+ except TableInformationException as tie:
+ logging.error(f"Cannot proceed: {tie}")
+ return dict()
+
+ all_results = dict()
+ for table in conf.tables:
+ map_data = get_partition_map(conf.dbcmd, table)
+ statistics = get_statistics(map_data["partitions"], conf.curtime, table)
+ all_results[table.name] = statistics
+
+ return all_results
+
+
+stats_parser = subparsers.add_parser("stats", help="get stats for partitions")
+stats_group = stats_parser.add_mutually_exclusive_group()
+stats_group.add_argument(
+ "--config", "-c", type=argparse.FileType("r"), help="Configuration YAML"
+)
+stats_group.add_argument("--table", "-t", type=SqlInput, nargs="+", help="table names")
+stats_parser.set_defaults(func=stats_cmd)
+
+
 def main():
  """
  Start here.

partitionmanager/cli_test.py

@@ -2,7 +2,7 @@
 import unittest
 from datetime import datetime, timezone
 from pathlib import Path
-from .cli import parser, partition_cmd
+from .cli import parser, partition_cmd, stats_cmd
 
 fake_exec = Path(__file__).absolute().parent.parent / "test_tools/fake_mariadb.sh"
 nonexistant_exec = fake_exec.parent / "not_real"
@@ -174,3 +174,15 @@ def test_partition_duration_seven_days(self):
 """
  )
  self.assertSequenceEqual(list(o), ["partitioned_last_week"])
+
+
+class TestStatsCmd(unittest.TestCase):
+ def test_stats(self):
+ args = parser.parse_args(
+ ["--mariadb", str(fake_exec), "stats", "--table", "partitioned_yesterday"]
+ )
+ r = stats_cmd(args)
+ self.assertEqual(r["partitioned_yesterday"]["partitions"], 2)
+ self.assertLess(r["partitioned_yesterday"]["time_since_last_partition"].days, 2)
+ self.assertNotIn("mean_partition_delta", r["partitioned_yesterday"])
+ self.assertNotIn("max_partition_delta", r["partitioned_yesterday"])

partitionmanager/stats.py

@@ -0,0 +1,64 @@
+import logging
+
+from datetime import timedelta
+from itertools import tee
+from .types import MaxValuePartition, Partition, UnexpectedPartitionException
+
+
+def pairwise(iterable):
+ """
+ iterable -> (s0,s1), (s1,s2), (s2, s3), ...
+ """
+ a, b = tee(iterable)
+ next(b, None)
+ return zip(a, b)
+
+
+def get_statistics(partitions, current_timestamp, table):
+ results = {"partitions": len(partitions)}
+
+ if not partitions:
+ return results
+
+ head_part = partitions[0]
+ tail_part = partitions[-1]
+
+ for p in partitions:
+ if not isinstance(p, Partition):
+ logging.warning(
+ f"{table} get_statistics called with a partition list "
+ + f"that included a non-Partition entry: {p}"
+ )
+ raise UnexpectedPartitionException(p)
+
+ if not isinstance(tail_part, MaxValuePartition):
+ logging.warning(
+ f"{table} get_statistics called with a partition list tail "
+ + f"that wasn't a MaxValuePartition: {p}"
+ )
+ raise UnexpectedPartitionException(tail_part)
+
+ if tail_part.timestamp():
+ results["time_since_last_partition"] = current_timestamp - tail_part.timestamp()
+
+ if head_part == tail_part:
+ return results
+
+ if head_part.timestamp() and tail_part.timestamp():
+ results["mean_partition_delta"] = (
+ tail_part.timestamp() - head_part.timestamp()
+ ) / (len(partitions) - 1)
+
+ max_d = timedelta()
+ for a, b in pairwise(partitions):
+ if not a.timestamp() or not b.timestamp():
+ logging.debug(f"{table} had partitions that aren't comparable: {a} and {b}")
+ continue
+ d = b.timestamp() - a.timestamp()
+ if d > max_d:
+ max_d = d
+
+ if max_d > timedelta():
+ results["max_partition_delta"] = max_d
+
+ return results

partitionmanager/stats_test.py

@@ -0,0 +1,69 @@
+import unittest
+from datetime import datetime, timedelta, timezone
+from .stats import get_statistics
+from .types import Table, MaxValuePartition, PositionPartition
+
+
+def mkPPart(name, *pos):
+ p = PositionPartition(name)
+ for x in pos:
+ p.add_position(x)
+ return p
+
+
+ts = datetime(1949, 1, 12, tzinfo=timezone.utc)
+
+
+class TestStatistics(unittest.TestCase):
+ def test_statistics_no_partitions(self):
+ s = get_statistics(list(), ts, Table("no_parts"))
+ self.assertEqual(s, {"partitions": 0})
+
+ def test_statistics_single_unnamed_partition(self):
+ s = get_statistics([MaxValuePartition("p_start", 1)], ts, Table("single_part"))
+ self.assertEqual(s, {"partitions": 1})
+
+ def test_statistics_single_partition(self):
+ s = get_statistics(
+ [MaxValuePartition("p_19480113", 1)], ts, Table("single_part")
+ )
+ self.assertEqual(
+ s, {"partitions": 1, "time_since_last_partition": timedelta(days=365)}
+ )
+
+ def test_statistics_two_partitions(self):
+ s = get_statistics(
+ [mkPPart("p_19480101", 42), MaxValuePartition("p_19490101", 1)],
+ ts,
+ Table("two_parts"),
+ )
+ self.assertEqual(
+ s,
+ {
+ "partitions": 2,
+ "time_since_last_partition": timedelta(days=11),
+ "mean_partition_delta": timedelta(days=366),
+ "max_partition_delta": timedelta(days=366),
+ },
+ )
+
+ def test_statistics_weekly_partitions_year(self):
+ parts = list()
+ base = datetime(2020, 5, 20, tzinfo=timezone.utc)
+ for w in range(0, 52):
+ partName = f"p_{base + timedelta(weeks=w):%Y%m%d}"
+ parts.append(mkPPart(partName, w * 1024))
+ parts.append(MaxValuePartition(f"p_{base + timedelta(weeks=52):%Y%m%d}", 1))
+
+ s = get_statistics(
+ parts, base + timedelta(weeks=54), Table("weekly_partitions_year_retention")
+ )
+ self.assertEqual(
+ s,
+ {
+ "partitions": 53,
+ "time_since_last_partition": timedelta(days=14),
+ "mean_partition_delta": timedelta(days=7),
+ "max_partition_delta": timedelta(days=7),
+ },
+ )

partitionmanager/types.py

@@ -1,7 +1,7 @@
 import abc
 import argparse
 import re
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from urllib.parse import urlparse
 
 
@@ -86,6 +86,17 @@ def num_columns(self):
  Return the number of columns this partition represents
  """
 
+ def timestamp(self):
+ """
+ Returns a datetime object representing this partition's
+ date, if the partition is of the form "p_YYYYMMDD", otherwise
+ returns None
+ """
+ try:
+ return datetime.strptime(self.name, "p_%Y%m%d").replace(tzinfo=timezone.utc)
+ except ValueError:
+ return None
+
  def __repr__(self):
  return f"{type(self).__name__}<{str(self)}>"
 

partitionmanager/types_test.py

@@ -1,7 +1,7 @@
 import argparse
 import unittest
-from datetime import timedelta
-from .types import Table, toSqlUrl, SqlInput, retention_from_dict
+from datetime import datetime, timedelta, timezone
+from .types import PositionPartition, retention_from_dict, SqlInput, Table, toSqlUrl
 
 
 class TestTypes(unittest.TestCase):
@@ -72,3 +72,14 @@ def test_table(self):
 
  r = retention_from_dict({"days": 30})
  self.assertEqual(timedelta(days=30), r)
+
+
+class TestPartition(unittest.TestCase):
+ def test_partition_timestamps(self):
+ self.assertIsNone(PositionPartition("").timestamp())
+ self.assertIsNone(PositionPartition("not_a_date").timestamp())
+ self.assertIsNone(PositionPartition("p_202012310130").timestamp())
+ self.assertEqual(
+ PositionPartition("p_20201231").timestamp(),
+ datetime(2020, 12, 31, tzinfo=timezone.utc),
+ )