# Rate calculation happens BEFORE Kafka: replay cannot corrupt counter deltas.
# State is per measurement + complete tag set; counters never mix across devices.
state = {}
COUNTERS = {
    "diskio": ["read_bytes", "write_bytes", "reads", "writes", "read_time", "write_time", "io_time"],
    "net": ["bytes_recv", "bytes_sent", "packets_recv", "packets_sent", "err_in", "err_out", "drop_in", "drop_out"],
    "kernel": ["context_switches", "interrupts", "processes_forked"],
    "swap": ["in", "out"],
}

def apply(metric):
    identity = metric.name + "|" + str(sorted(metric.tags.items()))
    if metric.name in COUNTERS:
        previous = state.get(identity)
        current = {}
        for field in COUNTERS[metric.name]:
            if field in metric.fields:
                current[field] = metric.fields[field]
        if previous != None and metric.time > previous[0]:
            elapsed = (metric.time - previous[0]) / 1000000000.0
            # Suppress gaps and resets: never turn them into false rate spikes.
            if elapsed <= 120.0:
                for field in current:
                    if field in previous[1] and current[field] >= previous[1][field]:
                        metric.fields[field + "_per_sec"] = (current[field] - previous[1][field]) / elapsed
            if metric.name == "diskio" and "io_time_per_sec" in metric.fields:
                metric.fields["io_busy_percent"] = min(100.0, metric.fields["io_time_per_sec"] / 10.0)
        if previous == None or metric.time > previous[0]:
            state[identity] = (metric.time, current)
    # Human-readable series distinguish same-named devices on different hosts.
    device = metric.tags.get("path", metric.tags.get("interface", metric.tags.get("name", metric.tags.get("cpu", "host"))))
    metric.tags["series"] = metric.tags.get("host", "unknown") + " / " + device
    if metric.name == "system" and metric.fields.get("n_cpus", 0) > 0:
        metric.fields["load1_per_cpu"] = metric.fields.get("load1", 0.0) / metric.fields["n_cpus"]
    return metric