sift_py.ingestion.config.telemetry

  1from __future__ import annotations
  2
  3import hashlib
  4from pathlib import Path
  5from typing import Any, Dict, List, Optional, cast
  6
  7from typing_extensions import Self
  8
  9from sift_py._internal.channel import channel_fqn
 10from sift_py.error import _component_deprecation_warning
 11from sift_py.ingestion.channel import (
 12    ChannelBitFieldElement,
 13    ChannelConfig,
 14    ChannelDataType,
 15    ChannelEnumType,
 16)
 17from sift_py.ingestion.config.yaml.load import (
 18    load_named_expression_modules,
 19    read_and_validate,
 20)
 21from sift_py.ingestion.config.yaml.spec import TelemetryConfigYamlSpec
 22from sift_py.ingestion.flow import FlowConfig
 23from sift_py.rule.config import (
 24    ExpressionChannelReference,
 25    ExpressionChannelReferenceChannelConfig,
 26    RuleAction,
 27    RuleActionAnnotationKind,
 28    RuleActionCreateDataReviewAnnotation,
 29    RuleActionCreatePhaseAnnotation,
 30    RuleConfig,
 31)
 32from sift_py.yaml.rule import RuleYamlSpec, load_rule_modules
 33
 34
 35class TelemetryConfig:
 36    """
 37    Configurations necessary to start ingestion.
 38    - `asset_name`: The name of the asset that you wish to telemeter data for.
 39    - `ingestion_client_key`: Optional string chosen by the user to uniquely identify this ingestion configuration. If this isn't
 40       supplied a sha256 hash will be used as the client key.
 41    - `flows`: A single flow can specify a single channel value or a set of channel values that are ingested together.
 42    - `organization_id`: ID of your organization in Sift. This field is only required if your user belongs to multiple organizations.
 43    - `rules`: Rules to evaluate during ingestion.
 44    """
 45
 46    asset_name: str
 47    ingestion_client_key: str
 48    organization_id: Optional[str]
 49    flows: List[FlowConfig]
 50    rules: List[RuleConfig]
 51    _ingestion_client_key_is_generated: bool
 52
 53    def __init__(
 54        self,
 55        asset_name: str,
 56        ingestion_client_key: Optional[str] = None,
 57        organization_id: Optional[str] = None,
 58        flows: List[FlowConfig] = [],
 59        rules: List[RuleConfig] = [],
 60    ):
 61        """
 62        Will raise a `TelemetryConfigValidationError` under the following conditions:
 63        - Multiple flows with the same name
 64        - Multiple rules with the same name
 65        - Identical channels in the same flow
 66        """
 67        self.__class__.validate_flows(flows)
 68        self.__class__.validate_rules(rules)
 69
 70        self.asset_name = asset_name
 71        self.organization_id = organization_id
 72        self.flows = flows
 73        self.rules = rules
 74
 75        if ingestion_client_key:
 76            self.ingestion_client_key = ingestion_client_key
 77            self._ingestion_client_key_is_generated = False
 78        else:
 79            self.ingestion_client_key = self.hash()
 80            self._ingestion_client_key_is_generated = True
 81
 82    @staticmethod
 83    def validate_rules(rules: List[RuleConfig]):
 84        """
 85        Ensure that there are no rules with identical names
 86        """
 87        seen_rule_names = set()
 88
 89        for rule in rules:
 90            if rule.name in seen_rule_names:
 91                raise TelemetryConfigValidationError(
 92                    f"Can't have two rules with identical names, '{rule.name}'."
 93                )
 94            seen_rule_names.add(rule.name)
 95
 96    @staticmethod
 97    def validate_flows(flows: List[FlowConfig]):
 98        """
 99        Ensures no duplicate channels and flows with the same name, otherwise raises a `TelemetryConfigValidationError` exception.
100        """
101        flow_names = set()
102
103        for flow in flows:
104            seen_channels = set()
105
106            if flow.name in flow_names:
107                raise TelemetryConfigValidationError(
108                    f"Can't have two flows with the same name, '{flow.name}'."
109                )
110
111            flow_names.add(flow.name)
112
113            for channel in flow.channels:
114                fqn = channel.fqn()
115
116                if fqn in seen_channels:
117                    raise TelemetryConfigValidationError(
118                        f"Can't have two identical channels, '{fqn}', in flow '{flow.name}'."
119                    )
120                else:
121                    seen_channels.add(fqn)
122
123    @classmethod
124    def try_from_yaml(
125        cls,
126        path: Path,
127        named_expression_modules: Optional[List[Path]] = None,
128        named_rule_modules: Optional[List[Path]] = None,
129    ) -> Self:
130        """
131        Initializes a telemetry config from a YAML file found at the provided `path` as well as optional
132        paths to named expression modules if named expressions are leveraged.
133        """
134
135        config_as_yaml = read_and_validate(path)
136
137        named_expressions = {}
138        rule_modules = []
139        if named_expression_modules is not None:
140            named_expressions = load_named_expression_modules(named_expression_modules)
141        if named_rule_modules is not None:
142            rule_modules = load_rule_modules(named_rule_modules)
143
144        return cls._from_yaml(config_as_yaml, named_expressions, rule_modules)
145
146    @classmethod
147    def _from_yaml(
148        cls,
149        config_as_yaml: TelemetryConfigYamlSpec,
150        named_expressions: Dict[str, str] = {},
151        rule_modules: List[RuleYamlSpec] = [],
152    ) -> Self:
153        rules = []
154        flows = []
155
156        for flow in config_as_yaml.get("flows", []):
157            channels = []
158
159            for channel in flow["channels"]:
160                data_type = cast(ChannelDataType, ChannelDataType.from_str(channel["data_type"]))
161
162                bit_field_elements = []
163                for bit_field_element in channel.get("bit_field_elements", []):
164                    bit_field_elements.append(
165                        ChannelBitFieldElement(
166                            name=bit_field_element["name"],
167                            index=bit_field_element["index"],
168                            bit_count=bit_field_element["bit_count"],
169                        )
170                    )
171
172                enum_types = []
173                for enum_type in channel.get("enum_types", []):
174                    enum_types.append(
175                        ChannelEnumType(
176                            name=enum_type["name"],
177                            key=enum_type["key"],
178                        )
179                    )
180                # NOTE: Component is deprecated, but warning raised in ChannelConfig init
181                channels.append(
182                    ChannelConfig(
183                        name=channel["name"],
184                        data_type=data_type,
185                        description=channel.get("description"),
186                        unit=channel.get("unit"),
187                        component=channel.get("component"),
188                        bit_field_elements=bit_field_elements,
189                        enum_types=enum_types,
190                    )
191                )
192
193            flows.append(
194                FlowConfig(
195                    name=flow["name"],
196                    channels=channels,
197                )
198            )
199
200        yaml_rules = config_as_yaml.get("rules", []) + rule_modules
201
202        for rule in yaml_rules:
203            action: Optional[RuleAction] = None
204            description: str = ""
205            annotation_type = RuleActionAnnotationKind.from_str(rule["type"])
206            tags = rule.get("tags")
207            description = rule.get("description", "")
208
209            action = RuleActionCreatePhaseAnnotation(tags)
210            if annotation_type == RuleActionAnnotationKind.REVIEW:
211                action = RuleActionCreateDataReviewAnnotation(
212                    assignee=rule.get("assignee"),
213                    tags=tags,
214                )
215
216            channel_references: List[
217                ExpressionChannelReference | ExpressionChannelReferenceChannelConfig
218            ] = []
219
220            for channel_reference in rule.get("channel_references", []):
221                for ref, val in channel_reference.items():
222                    name = val["name"]
223
224                    # NOTE: Component deprecated, kept for backwards compatibility
225                    component = val.get("component")
226                    if component:
227                        _component_deprecation_warning()
228
229                    channel_references.append(
230                        {
231                            "channel_reference": ref,
232                            "channel_identifier": channel_fqn(name, component),
233                        }
234                    )
235
236            expression = rule.get("expression", "")
237            rule_client_key = rule.get("rule_client_key", "")
238            if isinstance(expression, str):
239                rules.append(
240                    RuleConfig(
241                        name=rule["name"],
242                        description=description,
243                        expression=expression,
244                        action=action,
245                        rule_client_key=rule_client_key,
246                        channel_references=channel_references,
247                    )
248                )
249            else:
250                expression_name = cast(str, expression.get("name"))
251
252                expr = named_expressions.get(expression_name)
253
254                if expr is None:
255                    raise TelemetryConfigValidationError(
256                        f"Named expression '{expression_name}' could not be found. Make sure it was loaded in."
257                    )
258
259                sub_expressions = rule.get("sub_expressions", [])
260
261                sub_exprs: Dict[str, Any] = {}
262                for sub_expression in sub_expressions:
263                    for iden, value in sub_expression.items():
264                        sub_exprs[iden] = value
265
266                rules.append(
267                    RuleConfig(
268                        name=rule["name"],
269                        description=description,
270                        expression=expr,
271                        action=action,
272                        rule_client_key=rule_client_key,
273                        channel_references=channel_references,
274                        sub_expressions=sub_exprs,
275                    )
276                )
277
278        return cls(
279            asset_name=config_as_yaml["asset_name"],
280            ingestion_client_key=config_as_yaml.get("ingestion_client_key"),
281            organization_id=config_as_yaml.get("organization_id"),
282            rules=rules,
283            flows=flows,
284        )
285
286    def hash(self) -> str:
287        m = hashlib.sha256()
288        m.update(self.asset_name.encode())
289        for flow in sorted(self.flows, key=lambda f: f.name):
290            m.update(flow.name.encode())
291            # Do not sort channels in alphabetical order since order matters.
292            for channel in flow.channels:
293                m.update(channel.name.encode())
294                # Use api_format for data type since that should be consistent between languages.
295                m.update(channel.data_type.as_human_str(api_format=True).encode())
296                m.update((channel.description or "").encode())
297                # Deprecated.
298                m.update((channel.component or "").encode())
299                m.update((channel.unit or "").encode())
300                for bfe in sorted(channel.bit_field_elements, key=lambda bfe: bfe.index):
301                    m.update(bfe.name.encode())
302                    m.update(str(bfe.index).encode())
303                    m.update(str(bfe.bit_count).encode())
304                for enum in sorted(channel.enum_types, key=lambda et: et.key):
305                    m.update(str(enum.key).encode())
306                    m.update(enum.name.encode())
307
308        return m.hexdigest()
309
310
311class TelemetryConfigValidationError(Exception):
312    """
313    When the telemetry config has invalid properties
314    """
315
316    message: str
317
318    def __init__(self, message: str):
319        super().__init__(message)