333 Telemetry collection is configurable down to the Python module level

[tslominski]

The reason I hesitate to use custom pubsub is that I don’t think we have a clear, existing performance problem that justifies it. If the just-wrapping-logging approach becomes a performance problem, we’ll have clear evidence and a specific performance target to beat. Without that, we are signing up to maintain a custom framework for event dispatching. The way I’m thinking about it, we can always add new code to improve performance later, but we can get stuck with extra code and no performance improvement now.

I think the snapshots.{__name__} could still be relevant. Here’s the use cases I have in mind. The handlers that will replace the episode, epoch, and experiment summary metrics, will be consuming a telemetry stream of events. Those events are not snapshots. They’re pretty much metric events. For an experiment, we’ll always want these to be on. However, for visualizing an experiment with something like the live plotter, that handler would consume a snapshot stream that only gets turned on when we want live plotting. So the question is, how do we control this via logging configuration only.? This is where the snapshot prefix would be helpful. We could turn snapshots on and off per module (depending on what we want to visualize) while always having telemetry on, for example.

[AgentRev]

Did some changes:

Removed broker, clarified events and snapshots, separate loggers ·...

• Moved telemetry under frameworks, added logging levels within __init__.py;
• Removed the broker, TelemetryConsumer now subclasses logging.Handler;
• TelemetryEmitter now has two loggers with separate log levels, telemetry.{__name__} and snapshots.{__name__}, respectively emitted to via emit() and snapshot();
• Clarified schemas, events, snapshots: TelemetryEvent and TelemetrySnapshot are subclassed under TelemetrySchema;
• All mentions of schema “ID” are now schema “kind”;
• Consumers are now self-subscribing and “self-pumping” by default, i.e. they consume schemas immediately from within logging.Handler.emit() without queueing, but the queue-based pumpable variant is still available via __init__(deferred=True), which is required by threaded consumers.

No changes yet to the experiment / post-episode aggregation. That comes next.

[tslominski]

Thanks! This is looking good. I appreciate your updates and working with me to get this closer to what I’m hoping it will be. The shape of the TelemetryEmitter looks great.

Assuming that you’re looking for feedback …

One thing I ask you to keep in mind that’s not directly telemetry related but more of an overall Monty design direction is the distinction between the runtime and experiment domains. The distinction we are after is that someone could run Monty “for real” without ever knowing anything about experiments, episodes, epochs, etc. @DSchumacher has been doing a lot of work recently on splitting up runtime and experiments. An invariant to keep in mind is that runtime shall not import any experiment code, whereas experiment code can import the runtime code. We aren’t there yet with the existing code, but it’s something we can enforce on new code to make forward progress.

Where this comes up in telemetry is PostEpisodeTelemetry and PostEpisodeTelemetryConsumer. The words “post episode” alone mark those as exclusively experimental concerns. As such, these should end up in tbp.monty.experiment module somewhere (I suggest tbp.monty.experiment.telemetry). Since init_telemetry happens in MontyExperiment, which is … an experiment , then it can import experiment code without worry.

TelemetryEmitter looks like it’s in the right place .

Another point of caution of mixing experiment and runtime I’d highlight is in the TelemetrySchema. I don’t think that mode, episode, and step belong there, since they are experimental metadata (maybe step is OK, I’m not sure yet). Note that PostEpisodeTelemetryConsumer is managed by the experiment, which can tell it what mode, episode, and step it is on.

This raises the question of how epoch, episode, and step experiment data will be attached to events if they’re not set at the emit point.

---

I question if TelemetryStopEvent is needed. It feels to me like injecting a control plane message into the data plane.

---

I saw your note avoid levels >= 30 (logging.WARNING) (today I learned). I would assume that this behavior is only for the default logger, but neither telemetry nor snapshot loggers propagate to the default logger, so I’m wondering if this is a problem on the telemetry/snapshot side of things.

---

Are you able to set things up so that we would summon TelemetryEmitter in the same way we do logs?

import logging
import tbp.monty.frameworks.telemetry as tbp_telemetry
# or
from tbp.monty.frameworks.telemetry import getTelemetry

# ...

logger = logging.getLogger(__name__)
telemetry = tbp_telemetry.getTelemetry(__name__)
# or
telemetry = getTelemetry(__name__)

# ...

---

I still don’t fully understand the use case for deferred/pump.

Thank you again, the code is looking more and more like the interface I’m hoping for.

[AgentRev]

Thanks for the great feedback as always

@tslominski

I question if TelemetryStopEvent is needed. It feels to me like injecting a control plane message into the data plane.

Must be influences from my automotive work. On a CAN bus used in cars and robots, telemetry and control usually circulate across the same plane; 29-bit ID and 64-byte payload. I used the same philosophy here with TelemetryStopEvent, as an easy way to signal to threaded consumers to stop, since Queue.get() is a blocking call.

Maybe a Monty-centric event would be better, e.g. ExperimentEnd, in the same fashion as LearningModuleTerminalState. Consumers could then act upon it on their own terms, be it by stopping or otherwise. That way, it would remain solely a data plane.

@tslominski

I saw your note avoid levels >= 30 (logging.WARNING) (today I learned). I would assume that this behavior is only for the default logger, but neither telemetry nor snapshot loggers propagate to the default logger, so I’m wondering if this is a problem on the telemetry/snapshot side of things.

Ah, apparently this is from logging.lastResort when a logger has no handlers. So the comment I wrote is partially false. In my POC, I am emitting an ExperimentStatsTelemetry at end of episode, containing the same self.data forwarded to Wandb. Initially, I emitted it as a snapshot with level telemetry.TRACE = 30. It’s for future use, so its logger currently has no handlers, therefore I was seeing it getting printed to stderr. In that case, the solution to prevent levels >= 30 from printing is to call addHandler(logging.NullHandler()), avoiding logging.lastResort altogether.

@tslominski

Are you able to set things up so that we would summon TelemetryEmitter in the same way we do logs?

Certainly.

@tslominski

I still don’t fully understand the use case for deferred/pump.

Yeah, I too was starting to doubt of its relevance while writing the commit. The pump was for manually refreshing my live plotter POC at end of episode. However, a plotter consumer can just auto-refresh itself upon receiving any relevant schema, avoiding the need for pumping. Monty shouldn’t have to care about what its audience is doing.

I’ll remove deferred/pump, and the code relating to pump(continuous=True) will be transferred to ThreadedTelemetryConsumer. And, I think I’ll rename all mentions of “consumer” to “observer” (edit: “subscriber”) while I’m at it; that’ll be more accurate.

[tslominski]

Thank you, the updates sound good.

@AgentRev

Maybe a Monty-centric event would be better

I guess my question is, why is an event needed at all? I’m thinking the experiment is the coordinator of the entire run and has controls to all the software components. When the experiment is done, it can invoke whatever destructors are needed directly on the components it instantiated for the run.

[AgentRev]

@tslominski

I guess my question is, why is an event needed at all?

My reasoning was that if a thread calls Queue.get(), it gets blocked and will not react to external calls in the meantime, so an event would be an easy way to unblock it. However, on second thought, there’s also Queue.shutdown() that can be used for this, which the experiment could invoke via a stop() function or similar. Nonetheless, having state transition events in scientific tooling telemetry is relatively common, so I would still keep them on the table.

edit: Welp, Queue.shutdown() is Python 3.13+… So, it needs a stop event, or something like Queue.get(timeout=0.1) and self.stop_flag = threading.Event()

[AgentRev]

So, I’ve been working on the “running track of episode specific telemetry” side of things, and about your episode/correct example, I’m not sure if the code you posted is a suitable equivalent of the old code;

@tslominski

            case LearningModuleTerminalState(terminal_state=terminal_state):
                if terminal_state == "match":
                    self._episode_correct = self._episode_correct + 1

The episode/correct count is incremented by BasicGraphMatchingLogger.update_overall_stats(...) based on the value returned by the following code in logging_utils.get_graph_lm_episode_stats(...):

        elif primary_performance == "match":
            target_to_graph = lm.graph_id_to_target[lm.detected_object]
            if lm.primary_target in target_to_graph:
                primary_performance = "correct"

As you can see, episode/correct doesn’t depend solely on the terminal state, but on target detection as well.

My idea would be to instead emit some kinda GraphLMTerminalStats event from within GraphLM.set_individual_ts(...), e.g.

class GraphLMTerminalStats(TelemetryEvent):
    """Event produced by `GraphLM.set_individual_ts`."""

    KIND: Final[str] = "graph_lm_terminal_stats"

    lm_id: str
    """Learning module ID."""

    stats: dict
    """Output of `get_graph_lm_episode_stats`."""

Would that make sense? That way, the handler can keep a running track of values (e.g. via primary_performance) thru GraphLMTerminalStats events, which upon construction would directly use relevant logging_utils functions (involving a lot of direct calls to the learning module instances).

(I’m unsure if splitting it into more fine-grained events would be desirable, as it would likely incur a wider, high cost-to-benefit refactor of logging_utils functions.)

[tslominski]

The problem I see with the GraphLMTerminalStats approach is that it means the learning modules knows what the target is. This violates the (future, but we’re working towards it) Monty-does-not-know-about-experiments constraint. Monty should only emit events with information that it is allowed to know about. If telemetry emits events with experimental data in it, it means we gave Monty experimental data, which means we did not decouple Monty from experiment. Monty cannot emit any experiment statistics, because that implies it knows about experiments.

With this constraint in mind, what else would a learning module know? It knows its own object ID, so it could emit that as part of LearningModuleTerminalState:

event=LearningModuleTerminalState(
    kind="LearningModuleTerminalState",
    id=self.learning_module_id,
    terminal_state=TerminalState.MATCH,
    object_id=self.detected_object
)

We already log this, so the telemetry event would probably be the thing we log:

src/tbp/monty/frameworks/models/graph_matching.py

The handler (which is in the experiment, so it has experiment knowledge) knows what the target object ID was. So it should be able to match object_id with what it knows the episode target object ID is and discriminate between correct vs not correct.

[AgentRev]

@tslominski

The problem I see with the GraphLMTerminalStats approach is that it means the learning modules knows what the target is.

The main question looming over my head was “When does get_graph_lm_episode_stats gets called?” GraphLMTerminalStats was meant to accomplish that, collecting LM data that eventually gets forwarded to Wandb down the line. (While the function name contains the word “episode”, what it seems to output appear to actually be terminal stats rather than episode stats)

get_graph_lm_episode_stats only collects info that the learning module is aware about. Did you mean that target-related vars (lm.primary_target, lm.stepwise_target_object, etc.) are eventually getting removed from LMs?

At a quick glance, the dict keys that seem unrelated to the target are: num_steps, detected_location, detected_rotation, detected_scale, location_rel_body, detected_path, symmetry_evidence, individual_ts_reached_at_step, time. The experiment-unrelated ones among those would likely benefit from being emitted by GraphLM.set_individual_ts(...). Maybe all these could even be placed directly inside LearningModuleTerminalState? Or, would you prefer those to be collected from the LM instance on the handler side? (I was under the impression that handlers should not directly interact with LM instances, hence the event encapsulation)

[tslominski]

get_graph_lm_episode_stats has fields like “primary_performance” which a runtime LM can never populate as it doesn’t know the answer.

Yes, lm.primary_target etc are being removed.

@AgentRev

Or, would you prefer those to be collected from the LM instance on the handler side?

The way I’m thinking about it is that nothing can collect things from LM instances on the handler side. The handler’s only input would be telemetry that it consumes (it listens for LearningModuleTerminalState events and maybe others) and whatever the Experiment passes when it calls handler.post_episode(primary_target, ...). I imagine the side effect of handler.post_episode(...) call are the GraphLMTerminalStats events being emitted onto the telemetry stream, one for each episode.

I imagine there’s some sort of wandb handler wandb_handler downstream of the handler above, and it becomes a from-Monty-to-WandB transform without having to do any aggregation itself.

sequenceDiagram
    participant telemetry@{ "type" : "queue" }
    telemetry->>handler: LearningModuleTerminalState
    telemetry->>handler: LearningModuleTerminalState
    telemetry->>handler: LearningModuleTerminalState
    telemetry->>handler: LearningModuleTerminalState
    participant experiment@{ "type" : "entity" }
    experiment->>+handler: post_episode(primary_target, ...)
    handler->>-telemetry: GraphLMTerminalStats
    telemetry->>+wandb_handler: GraphLMTerminalStats
    participant wandb@{ "type" : "boundary" }
    wandb_handler->>-wandb: wandb stuff

I would also question the “terminal” in LM state that we currently have in the code. That’s experiment episode language leaking into the LM. There’s probably something more appropriate in the future we can use like “recognized” or something along those lines (but I don’t think addressing it is important right now).

[AgentRev]

I seem to be going around in a bit of a circle here, your input would be appreciated.

graph_matching_loggers calls get_stats_per_lm, which itself calls (and passes down the LM instances to) get_graph_lm_episode_stats, add_evidence_lm_episode_stats, add_pose_lm_episode_stats, and add_policy_episode_stats, which themselves collect values from LMs.

Among those, apart from the primary_target stuff, get_graph_lm_episode_stats reads quite a few things:

• lm.buffer object contents (get_num_matching_steps(), get_current_location(), on_object list, stats dict);
• lm.graph_id_to_target dict values (depending on the content of lm.buffer.stats);
• lm.target_to_graph_id dict values;
• lm.detected_pose;
• lm.symmetry_evidence.

The collecting process itself varies depending on primary_performance and primary target detection; they’re entangled.

I’m hesitant on how to reconcile all of this with “nothing can collect things from LM instances on the handler side”. My intuition would be to just call get_stats_per_lm on the experiment side and pass the resulting dict via telemetry event to the handler. But I feel this wouldn’t be aligned what you’re looking for.

At the same time, “yeeting” a bunch of those LM infos into telemetry events and letting the handler(s) pick them apart does not seem like the right way either. It would essentially be recreating a fragmented, event-based, partial “carbon copy” of the LMs, and complexify the code quite a bit.

What’s your take on this? Do you have a preference between either of these two approaches, or is there another alternative eluding me?

[tslominski]

I think “yeeting” the LM infos into telemetry is the approach that aligns with what I outlined. Recreating a fragmented, event-based, partial “carbon copy” of the LMs is what handling telemetry streams is. As to complexity, currently, the complexity lives in the runtime. My goal for the telemetry approach is to simplify the runtime and push the complexity into the experiment framework.

The docstring for get_stats_per_lm is "Loop through lms and get stats." So, the way I translate that is, "loop through lms" turns into each LM having to, at some point(s) in the episode lifecycle, emit the requisite data so that the handler can consume those events and do the "get stats" part.

The get_graph_lm_episode_stats looks to me like the contents LearningModuleObjectRecognized event (this is me trying not to use “terminal state” in a name).

All the things conditional on primary_performance would be the different contents that can be inside a LearningModuleObjectRecognized event. It is a bit muddled whether there is one event type or multiple because the shape of the event changes on runtime outcomes (recognized / not recognized) as well as configuration parameters (different types of learning modules).

Maybe one hint that I see that might be helpful is that, if the handler does not see a LearningModuleObjectRecognized event from a learning module when the experiment tells it to post_episode, that means that the particular LM for which the event is missing did not recognized the object. To handle if len(possible_matches) == 0: branch, the LMs need to emit possible matches events at each step. To handle elif lm.primary_target == "no_label":, the handler gets the primary target in post_episode from experiment. To handle elif primary_performance == "match": branch, this comes from LM emitting LearningModuleObjectRecognized event. To handle elif primary_performance == "time_out": branch, this means the LM never emitted LearningModuleObjectRecognized event during this episode. All the other control paths hint at additional info and events that need to come from the LM.

I think you see the solution, it’s just complex, and I agree . We are moving all this complexity to a different place, so I think it’s going to be a lot.

I keep saying “moving,” but in order to get this integrated in a continuous fashion, this is all new functionality. Only once the new functionality is working, will we be able to delete the old functionality and complete the “move.”