Repository: risingwavelabs/risingwave
File: grafana/dashboard/dev/streaming_fragments.py (Python)

Code before:
    actor_level_filter = "actor_id!=''"
    panels = outer_panels.sub_panel()
    actor_count_expr = (
        f"clamp_min(sum({metric('stream_actor_count')}) by (fragment_id), 1)"
    )
    poll_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_poll_duration')}[$__rate_interval])) by (fragment_id)"
    )
    poll_duration_expr = (
        f"{poll_duration_rate_expr} / on(fragment_id) {actor_count_expr} / 1000000000"
    )
    idle_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_idle_duration')}[$__rate_interval])) by (fragment_id)"
    )
    idle_duration_expr = (
        f"{idle_duration_rate_expr} / on(fragment_id) {actor_count_expr} / 1000000000"
    )
    scheduled_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_scheduled_duration')}[$__rate_interval])) by (fragment_id)"
    )
    scheduled_duration_expr = (
        f"{scheduled_duration_rate_expr} / on(fragment_id) {actor_count_expr} / 1000000000"
    )
    zero_fragment_expr = f"0 * sum({metric('stream_actor_count')}) by (fragment_id)"
    def _coalesce_fragment(expr: str) -> str:
        return f"(({expr}) or on(fragment_id) {zero_fragment_expr})"

    executor_cache_usage_expr = (
        f"sum("
        f"  ({metric('stream_memory_usage')})"
        f"  * on (table_id) group_left(fragment_id)"
        f"    {metric('table_info')}"
        f") by (fragment_id)"
    )
    shared_buffer_usage_expr = (
        f"sum({metric('state_store_per_table_imm_size')}) by (fragment_id)"
    )
    kv_log_store_buffer_usage_expr = (
        f"sum({metric('kv_log_store_buffer_memory_bytes')}"
        f"  * on(actor_id) group_left(fragment_id) {metric('actor_info')})"
        f" by (fragment_id)"
    )
    sync_kv_log_store_buffer_usage_expr = (
        f"sum({metric('sync_kv_log_store_buffer_memory_bytes')}) by (fragment_id)"
    )
    total_memory_usage_expr = " + ".join(
        map(
            _coalesce_fragment,
            [
                executor_cache_usage_expr,
                shared_buffer_usage_expr,

Review comment (suggestion):
The variable name `zero_fragment_expr` is ambiguous. Consider renaming to `zero_for_all_fragments_expr` or `fragment_zero_baseline_expr` to better convey that it creates a zero value for all fragment IDs.
```suggestion
    zero_for_all_fragments_expr = f"0 * sum({metric('stream_actor_count')}) by (fragment_id)"
    def _coalesce_fragment(expr: str) -> str:
        return f"(({expr}) or on(fragment_id) {zero_for_all_fragments_expr})"
```

Code after:
    """Compute the peak per-actor rate (normalized to seconds) and attach `id`."""
    return (
        f"label_replace("
        f"(max_over_time((({per_actor_expr}) / 1000000000)[$__range:$__interval])), "
        f"'id', '$1', 'fragment_id', '(.+)')"
    )

def _kv_log_store_buffer_usage_by_fragment_expr() -> str:
    return (
        f"sum({metric('kv_log_store_buffer_memory_bytes')}"
        f"  * on(actor_id) group_left(fragment_id) {metric('actor_info')})"
        f" by (fragment_id)"
    )

def _sync_kv_log_store_buffer_usage_by_fragment_expr() -> str:
    return f"sum({metric('sync_kv_log_store_buffer_memory_bytes')}) by (fragment_id)"

@section
def _(outer_panels: Panels):
    # The actor_id can be masked due to metrics level settings.
    # We use this filter to suppress the actor-level panels if applicable.
    actor_level_filter = "actor_id!=''"
    panels = outer_panels.sub_panel()
    actor_count_expr = (
        f"clamp_min(sum({metric('stream_actor_count')}) by (fragment_id), 1)"
    )
    poll_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_poll_duration')}[$__rate_interval])) by (fragment_id)"
    )
    poll_duration_expr = (
        f"{poll_duration_rate_expr} / on(fragment_id) {actor_count_expr}"
    )
    idle_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_idle_duration')}[$__rate_interval])) by (fragment_id)"
    )
    idle_duration_expr = (
        f"{idle_duration_rate_expr} / on(fragment_id) {actor_count_expr}"
    )
    scheduled_duration_rate_expr = (
        f"sum(rate({metric('stream_actor_scheduled_duration')}[$__rate_interval])) by (fragment_id)"
    )
    scheduled_duration_expr = (
        f"{scheduled_duration_rate_expr} / on(fragment_id) {actor_count_expr}"
    )
    zero_for_all_fragments_expr = f"0 * sum({metric('stream_actor_count')}) by (fragment_id)"
    def _coalesce_fragment(expr: str) -> str:
        return f"(({expr}) or on(fragment_id) {zero_for_all_fragments_expr})"

    executor_cache_usage_expr = (
        f"sum("
        f"  ({metric('stream_memory_usage')})"