Function save

chirpstack/src/storage/metrics.rs:92–166

(name: &str, record: &Record, aggregations: &[Aggregation])

}

pub async fn save(name: &str, record: &Record, aggregations: &[Aggregation]) -> Result<()> {
    if record.metrics.is_empty() {
        return Ok(());
    }

    let mut pipe = redis::pipe();
    pipe.atomic();

    for a in aggregations {
        let ttl = get_ttl(*a);

        let ts: NaiveDateTime = match a {
            Aggregation::MINUTE => {
                NaiveDate::from_ymd_opt(record.time.year(), record.time.month(), record.time.day())
                    .ok_or_else(|| anyhow!("Invalid date"))?
                    .and_hms_opt(record.time.hour(), record.time.minute(), 0)
                    .ok_or_else(|| anyhow!("Invalid time"))?
            }
            Aggregation::HOUR => {
                NaiveDate::from_ymd_opt(record.time.year(), record.time.month(), record.time.day())
                    .ok_or_else(|| anyhow!("Invalid date"))?
                    .and_hms_opt(record.time.hour(), 0, 0)
                    .ok_or_else(|| anyhow!("Invalid time"))?
            }
            Aggregation::DAY => {
                NaiveDate::from_ymd_opt(record.time.year(), record.time.month(), record.time.day())
                    .ok_or_else(|| anyhow!("Invalid date"))?
                    .and_hms_opt(0, 0, 0)
                    .ok_or_else(|| anyhow!("Invalid time"))?
            }
            Aggregation::MONTH => {
                NaiveDate::from_ymd_opt(record.time.year(), record.time.month(), 1)
                    .ok_or_else(|| anyhow!("Invalid date"))?
                    .and_hms_opt(0, 0, 0)
                    .ok_or_else(|| anyhow!("Invalid time"))?
            }
        };

        let key = get_key(name, *a, ts);

        for (k, v) in &record.metrics {
            // Passing a reference to hincr will return a runtime error.
            let k = k.clone();
            let v = *v;

            match record.kind {
                Kind::COUNTER => {
                    pipe.cmd("HSET").arg(&key).arg(k).arg(v).ignore();
                }
                Kind::ABSOLUTE => {
                    pipe.cmd("HINCRBYFLOAT").arg(&key).arg(k).arg(v).ignore();
                }
                Kind::GAUGE => {
                    pipe.cmd("HINCRBYFLOAT")
                        .arg(&key)
                        .arg(format!("_{}_count", k))
                        .arg(1.0)
                        .ignore();