Implement a data sourcing actor

[tiyash-basu-frequenz]

What's needed?

We need to implement an actor that can expose data from different sources to the downstream layers.

Component data can come from different sources, e.g., the Microgrid API, the Reporting API, and parquet files in local storage.
The data format in these sources can be different.

There may be mutiple downstream actors, each of them reading from multiple sources.
If they all have mechanisms to convert data into one common format for further consumption, then:

• they would each need to understand how all the different data sources work, causing high coupling, and
• they would very likely run similar conversions, wasting CPU cycles

Proposed solution

The solution would be to create an actor that

• can read multiple data sources, e.g., the Microgrid API, and parquet files in local storage,
• can expose data in a common format that abstracts the underlying data source away,
• can switch between input data streams seamlessly, based upon well-defined conditions (see below for details).

Note that in the first version (v1), this actor needs to consume data from the following sources only:

• Microgrid API
• parquet files in local storage

In order for it to identify the data channel and the data source(s), the requests to the actor should consists of the following parameters:
a. the component ID
b. the metric ID
c. the start timestamp, or the timestamp from which data should be read

If the start timestamp is in past, then the local parquet files should be read first. Once there is no more data in the parquet files, then data from the live data source (Microgrid API) should be streamed.

The top-level building blocks of the actor could be a set of classes as follows
(Note that these are just pseudocoded ideas to specify requirements. The actual implementation may vary):

# This is the class that uniquely represents a metric channel.
@dataclass
class ChannelKey:
    # An enum or a unique ID representing the metric.
    metric_id: str

    # The component ID
    component_id: int

    # The time from which data should be consumed.
    start_ts: datetime

See ComponentMetricKey in issue #5 for more details on ChannelKey.

# This is the class that consumes and forwards data for one given channel.
# The actor can have a collection of the objects of this class.
class DataIngestor:
  _channel: Option[BroadcastChannel[float]]

  def __init_(self):
    # Initialize `_channel` as `None`
    self._channel = None
    ...

  def get_channel(self, data_key: ChannelKey) -> BroadcastChannel[float]:
    # This method is key here.
    #
    # Consume data from different sources, and stream them downstream via an
    # output channel. The user has to input the details of which
    # metric has to be consumed and from what time. The user gets a channel with
    # the desired data in return.
    #
    # If `data_key.start_ts` is `now()`, then read directly from the
    # Microgrid API.
    #
    # If `data_key.start_ts` is in the past, e.g., `now() - 30m`, then read the
    # parquet files until there is no more data there. Then read from the
    # microgrid API.
    #
    # If `data_key.start_ts` is in the future, e.g., `now() + 30m`, then wait
    # until the given timestamp is reached. Then read from the microgrid API.

    # Return `self._channel` if it already exists, otherwise create a new one.
    ...

The actor itself could be defined as follows:

@dataclass
class DataRequestKey:
  # The details of the metric to read data for, and the timesamo from which the
  # data should be read.
  data_key: ChannelKey

  # Can have more elements as required.

class DataSourcingActor:
  # The idea here is to store previously returned channels, so that we reuse them
  # insted of creating new ones every time. The data structure can be a `dict`, or
  # can be some other custom data structure, depending upon the following
  # requirements:
  #
  # Two `ChannelKey` objects with `start_ts >= now()` are
  # effectively the same, since they consume live data from the Microgrid API.
  #
  # However, two `ChannelKey` objects that need historical data from the
  # same `start_ts` timestamp may not be the same. Assuming the two
  # `ChannelKey` objects are received at times 00:00:00 and 01:00:00,
  # then the 2nd one will still require to start from the the first historical
  # data sample, whereas the 1st one could be several data samples ahead or
  # reading live data by then.
  _data_ingestors: Dict[DataRequestKey, DataIngestor]

  def __init__(self):
    # Initialize the `_data_ingestors` object.
    self._data_ingestors = {}
    ...

  def get_channel(self, data_key: ChannelKey):
    data_req_key = DataRequestKey(data_key)

    # If a `DataIngestor`1 object has not yet been created, then create it.
    if not self._data_ingestors.contains(data_req_key):
      self._data_ingestors[data_req_key] = DataIngestor()

    # Return the channel from the `DataNgestor` instance.
    return self._data_ingestors[data_req_key].get_channel()

Use cases

No response

Alternatives and workarounds

No response

Additional context

No response

[thomas-nicolai-frequenz]

@sahas-subramanian-frequenz can you please add the details with regards to reading from the parquet files and the buffers we talked about so that we get a fair distribution of load across different reading processes and smartly yield.

[thomas-nicolai-frequenz]

First problem is reading from parquet needs to happen in small chunks if we are reading things in parallel to avoid one reading "process" to be faster and to block all others. So this needs to be non-blocking as possible and avoiding feeding one channel faster than others.

Second issue is if the a buffer is full the reading process from parquet might pause but this is not possible with the Microgrid API. There we have two choices: (1) throw away records (2) push them to disk and consumer later if buffer gets empty again. Point 2 has the problem that for realtime cases like grid fuse protection etc. you are probably not interest in the oldest records from disk but these might be needed for e.g. building the averages for the past 15min. So we would need to make sure that priority does have current data while older data has lower priority or we need to be able to define in the channel properties that in one channel there is no disk overflow and in others there are. Thoughts?

All problems are also marked with a red exclamation mark in the slide:

[thomas-nicolai-frequenz]

properties that in one channel there is no disk overflow and in others there are

Assuming that use-case actors know best which data can be thrown away like the ev actor or circuit breaker while others like pv optimization need it the argument might be to support it in the channel properties. I'm just not sure if this is v1 or v2. Feels more like v2 imho but open to discuss.

[thomas-nicolai-frequenz]

I think ordering is cheap. A simple (maybe slightly adjusted) bubble sort on the time stamps should be very efficient given the invariants we have in the buffer, i.e. it's sorted after each operation. My bet is that such an algorithm has an average constant runtime.

Some comment from @matthias-wende-frequenz with regards to sorting incoming samples in the LogicalMeter/*Pool actors if they are being out of order. If there is no problem in a single process environment fine but I don't think we should bet on that.

[leandro-lucarella-frequenz]

A few questions:

1. If the start timestamp is in past, then the local parquet files should be read first. Once there is no more data in the parquet files, then data from the live data source (Microgrid API) should be streamed.

   When does reading from the Report API comes into play?

2. Silly bike-shedding: I think we can use DataSource instead of DataIngestor.

3. def get_channel(self, data_key: ChannelKey) -> BroadcastChannel[float]:

   Shouldn't this be get_receiver(...) -> Receiver[float]: instead?

4. If data_key.start_ts is now(), then read directly from the
   Microgrid API.

   I know is pseudocode but I would say None instead of now() as we can't possible compare now() == now() as they will be always different :D

5. If data_key.start_ts is in the future, e.g., now() + 30m, then wait
   until the given timestamp is reached. Then read from the microgrid API.

   Do we really have an use case for this?

6. So, if we need to have multiple channels per (component_id, metric_id, start_time), then we need to add some sort of UUID to the channel name. If that is the case, then we don't have anymore a way to infer a channel name based on the data requested to the data source, so I think we need to have this in mind in general.

7. The DataSourcingActor is missing the run() method. That method will have to do the conversion of data (in the microgrid API converting a ComponentData object into individual float values). But I think this conversion should be done in the DataSource class. Even more, I think we should probably have a MicrogridDataSource, LocalStorageDataSource, etc. classes that are in charge of the particular conversions, so we can easily extend it to add more sources without making the DataSource a spaghetti.

8. I think DataSourcingActor.get_channel() shouldn't be a method but a message passed via a control channel to the actor (or the method could be preserved to call it at initialization-time). If I didn't understand everything wrong, that was the whole point of the ChannelRegistry, to be able to dynamically request for channels?

   So we should probably have a DataSourcingActor.SubscribeRequestMessage like:

   @dataclass
   class SubscribeRequestMessage:
       metric_id: str
       component_id: int
       start_ts: datetime

   And a DataSourcingActor.SubscribeResponseMessage like:

   @dataclass
   SubscribeResponseMessage:
       channel_name: str

   Then the actor making the request can go to the ChannelRegistry and do .get_receiver(response_message.channel_name) to get the receiver.