Merge pull request #148 from SUSE-Enceladus/data-archive

Add metering archive

README.md

@@ -33,6 +33,15 @@ layer. These include **`setup_adapter`**, **`load_defaults`**, and
 **`get_usage_data`**. Details about these hookspecs are found in
 [hookspecs.py](https://github.com/SUSE-Enceladus/csp-billing-adapter/blob/main/csp_billing_adapter/hookspecs.py).
 
+### Archive layer
+
+There are three hookspecs for retrieving and persisting the metering
+archive to stateful storage. A plugin implementing the hooks for
+the archive layer is required for the archive functionality to work.
+The hooks include **`get_archive_location`**, **`get_metering_archive`**,
+**`save_metering_archive`**. Details can be found in
+[archive_hookspecs.py](https://github.com/SUSE-Enceladus/csp-billing-adapter/blob/data-archive/csp_billing_adapter/archive_hookspecs.py).
+
 ## Configuration
 
 The adapter is configured via a YAML file that is accesible to the
@@ -47,6 +56,8 @@ api: THE_PATH_WHERE_WE_GET_THE_UNIT_INFORMATION
 billing_interval: monthly|hourly
 query_interval: Time in seconds
 product_code: TBD
+archive_retention_period: 6
+archive_bytes_limit: 1048576
 usage_metrics:
   metric_1:
     usage_aggregation: maximum
@@ -185,6 +196,42 @@ timestamp that denotes when the usage record was reported and
 **dimensions:** A dictionary containing the dimension names and units
 billed in the last metering.
 
+### Archive
+
+This is the metering and usage records archive. It contains a list of
+metering records with the relevent usage data, billing status, and
+timestamps. The format is as follows:
+
+```
+[
+  {
+        'billing_time': 'string(date timestamp)',
+        'billing_status': {
+            '{usage_key}': {
+                'record_id': string,
+                'status': 'succeeded'
+            }
+        },
+        'billed_usage': {
+            '{usage_key}': int
+        },
+        'usage_records': [
+            {
+                '{usage_metric}': int,
+                'reporting_time': 'string(date timestamp)',
+                'base_product': 'cpe:/o:suse:product:v1.2.3'
+            },
+            ...
+        ]
+    }
+]
+```
+
+**billing_time:** A timestamp denoting when the metering record was submitted.
+**billing_status:** A dictionary of the dimensions billed and billing status.
+**billed_usage:** A dicitonary of the usage incurred during the billing cycle per dimension.
+**usage_records:** A list of all the usage records logged during the billing period.
+
 ## Service
 
 The adapter service runs continuously based on the query interval which is

csp_billing_adapter/adapter.py

@@ -49,6 +49,7 @@
     csp_hookspecs,
     hookspecs,
     storage_hookspecs,
+    archive_hookspecs,
     hookimpls
 )
 
@@ -71,6 +72,7 @@ def get_plugin_manager() -> pluggy.PluginManager:
     pm.add_hookspecs(hookspecs)
     pm.add_hookspecs(csp_hookspecs)
     pm.add_hookspecs(storage_hookspecs)
+    pm.add_hookspecs(archive_hookspecs)
     pm.register(hookimpls)
     pm.load_setuptools_entrypoints('csp_billing_adapter')
     return pm
@@ -164,7 +166,11 @@ def initial_adapter_setup(
             func_name="hook.get_account_info"
         )
 
-        csp_config = create_csp_config(config, account_info)
+        csp_config = create_csp_config(
+            config,
+            account_info,
+            hook.get_archive_location()
+        )
 
     # Update csp-config with latest plugin versions
     versions = hook.get_version()

csp_billing_adapter/archive.py

@@ -0,0 +1,129 @@
+#
+# Copyright 2024 SUSE LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Utility functions for handling a rolling dictionary archive."""
+
+import functools
+import logging
+import json
+
+from csp_billing_adapter.config import Config
+from csp_billing_adapter.utils import retry_on_exception
+
+log = logging.getLogger('CSPBillingAdapter')
+
+
+def append_metering_records(
+    archive: list,
+    billing_record: dict,
+    max_length: int,
+    max_bytes: int
+) -> list:
+    """
+    Append usage and metering records to the archive
+
+    If archive is larger than max length drop the oldest
+    record. If the archive is larger than the max bytes
+    limit trim it until it satisfies the limit.
+
+    :param archive:
+        The list of meterings and usage records to append to.
+    :param billing_record:
+        The dictionary containing the most recent
+        metering and usage records to be archived.
+    :param max_length:
+        The max size of the archive list.
+    :param max_bytes:
+        The max size in bytes of the archive.
+    :return:
+        The archive of meterings and usage records with the
+        billing_record appended. If archive ends up greater
+        than max lengh or max bytes the archive is trimmed
+        as necessary to satisfy both max_length and
+        max_bytes.
+    """
+    archive.append(billing_record)
+
+    if len(archive) > max_length:
+        archive = archive[1:]
+
+    if max_bytes > 1:
+        # Treat 0 and 1 the same. Disable max bytes option.
+        # This prevents infitite loop when value is 1 since
+        # empty list is 2 bytes.
+        while True:
+            # Trim archive until it is smaller than max bytes
+            archive_size = len(bytes(json.dumps(archive), 'utf-8'))
+
+            if archive_size > max_bytes:
+                archive = archive[1:]
+            else:
+                break
+
+    return archive
+
+
+def archive_record(
+    hook,
+    config: Config,
+    billing_record: dict
+) -> None:
+    """
+    :param hook:
+        The Pluggy plugin manager hook that will be
+        used to call the meter_billing operation.
+    :param config:
+        The configuration specifying the metrics that
+        need to be processed in the usage records list.
+    :param billing_record:
+        The dictionary containing the most recent
+        metering and usage records to be archived.
+    """
+    retention_period = config.archive_retention_period
+    bytes_limit = config.archive_bytes_limit
+
+    if retention_period < 1 or bytes_limit in (0, 1):
+        # Archive feature has been disabled, do nothing
+        return
+
+    archive = retry_on_exception(
+        functools.partial(
+            hook.get_metering_archive,
+            config=config,
+        ),
+        logger=log,
+        func_name="hook.get_metering_archive"
+    )
+
+    if archive is None:
+        archive = []
+
+    archive = append_metering_records(
+        archive,
+        billing_record,
+        retention_period,
+        bytes_limit
+    )
+
+    retry_on_exception(
+        functools.partial(
+            hook.save_metering_archive,
+            config=config,
+            archive_data=archive
+        ),
+        logger=log,
+        func_name="hook.save_metering_archive"
+    )

csp_billing_adapter/archive_hookspecs.py

@@ -0,0 +1,54 @@
+#
+# Copyright 2024 SUSE LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Pluggy hook interface specifications for archival storage related functionality
+"""
+
+import pluggy
+
+from csp_billing_adapter.config import Config
+
+hookspec = pluggy.HookspecMarker('csp_billing_adapter')
+
+
+@hookspec(firstresult=True)
+def get_archive_location() -> str:
+    """
+    Returns the location of the archive data storage
+    """
+
+
+@hookspec(firstresult=True)
+def get_metering_archive(config: Config) -> list:
+    """
+    Retrieves the archive data from stateful storage
+
+    :param config: The application configuration dictionary
+    :return: Return a list of the archive data which contains a history
+             of recent data usage and meterings. The length of this data
+             is determined by application config.
+    """
+
+
+@hookspec(firstresult=True)
+def save_metering_archive(config: Config, archive_data: list) -> None:
+    """
+    Saves the archive data to stateful storage
+
+    :param config: The application configuration dictionary
+    :param archive_data: A list of usage data and meterings
+    """

csp_billing_adapter/bill_utils.py

@@ -39,6 +39,7 @@
     string_to_date
 )
 from csp_billing_adapter.config import Config
+from csp_billing_adapter.archive import archive_record
 
 log = logging.getLogger('CSPBillingAdapter')
 
@@ -655,3 +656,21 @@ def process_metering(
             )
             csp_config['usage'] = billable_usage
             csp_config['last_billed'] = metering_time
+
+            # Save last usage and metering records to archive
+            billing_record = {
+                'billing_time': metering_time,
+                'billing_status': billing_status,
+                'billed_usage': billed_dimensions,
+                'usage_records': billable_records
+            }
+
+            try:
+                archive_record(
+                    hook,
+                    config,
+                    billing_record
+                )
+            except Exception as error:
+                # Non-fatal error is only logged
+                log.exception(error)

csp_billing_adapter/csp_config.py

@@ -34,7 +34,8 @@
 
 def create_csp_config(
     config: Config,
-    account_info: dict
+    account_info: dict,
+    archive_location: str
 ) -> dict:
     """
     Initialize the csp_config data store.
@@ -44,6 +45,8 @@ def create_csp_config(
     :param account_info:
         A dictionary containing CSP account info that is
         added to the csp_config.
+    :param archive_location:
+        The data archive location.
     """
     now = get_now()
     expire = date_to_string(get_date_delta(now, config.reporting_interval))
@@ -53,6 +56,7 @@ def create_csp_config(
         'timestamp': date_to_string(now),
         'expire': expire,
         'customer_csp_data': account_info,
+        'archive_location': archive_location,
         'errors': []
     }
 

csp_billing_adapter/memory_archive.py

@@ -0,0 +1,45 @@
+#
+# Copyright 2024 SUSE LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""In-memory archive plugin implementation."""
+
+import logging
+
+import csp_billing_adapter
+
+from csp_billing_adapter.config import Config
+
+memory_archive = []
+
+log = logging.getLogger('CSPBillingAdapter')
+
+
+@csp_billing_adapter.hookimpl(trylast=True)
+def get_archive_location():
+    """Retrieve archive location."""
+    return '/tmp/fake_archive.json'
+
+
+@csp_billing_adapter.hookimpl(trylast=True)
+def get_metering_archive(config: Config):
+    return memory_archive.copy()
+
+
+@csp_billing_adapter.hookimpl(trylast=True)
+def save_metering_archive(config: Config, archive_data: list):
+    global memory_archive
+
+    memory_archive = archive_data

examples/README.md

@@ -30,6 +30,12 @@ reporting_api_is_cumulative:
 reporting_interval: 
     Sets the time in seconds when the csp-billing-adapter reports to the CSP API. The values the csp-billing-adapter reports are determined by a combination of the settings for `reporting_api_is_cumulative` and `billing_interval`.
 
+archive_retention_period:
+    Sets the time in billing cycles that meterings and usage records are retained in the data archive. For example when the billing period is in months a value of 6 will retain 6 months of records. If the value is set to 0 the archive functionality will be disabled.
+
+archive_bytes_limit:
+    Sets the max size in bytes for the metering archive. The archive will be trimmed to stay below this limit. To disable the bytes limit set the value to -1. To disable the archive functionality set the value to 0.
+
 usage_metrics: 
 
   {metric name}: