nsmcp::AccessCardinalityStore method init (protected)

 <instance of nsmcp::AccessCardinalityStore> init

Defined in /usr/local/ns/tcl/nsmcp/log-traffic-aggregate.tcl

Testcases:

No testcase defined.

Source code:

set :data ${:checkpoint}
set fresh [expr {${:data} eq {}}]
if {$fresh} {
    set :data [dict create backend sqlite status active reason {} path {} owner [ns_crypto::randombytes -encoding hex 16]  pool [dict get ${:limits} aggregatePool] maxBytes [dict get ${:limits} aggregateBytes]  peerOmitted 0 agentOmitted 0 peakBufferedKeys 0 labelTruncatedRecords 0 labelRedactedRecords 0 diskBytes 0]
}
# Old checkpoints with a spool predate explicit schema-ready tracking.
set defaults [dict create schemaReady [expr {!$fresh && [dict get ${:data} path] ne {}}]  poolWaitCount 0 poolWaitMilliseconds 0 poolWaitLimitMilliseconds  [:dict getdef ${:limits} aggregateWaitMs 30000] failure {}]
set :data [dict merge $defaults ${:data}]
if {[dict get ${:data} status] in {unavailable limited} && !${:readResults}} {
    return
}
if {${:readResults} && ![dict get ${:data} schemaReady] && [dict get ${:data} status] eq "unavailable"} {
    return
}
if {${:readResults} && [dict get ${:data} failure] ne {}
    && [dict get ${:data} failure class] in {invalid-database corrupt-database spool-missing owner-mismatch wrong-driver}} {
    return
}
set phase acquire-pool
try {
    set remaining [expr {[dict get ${:data} poolWaitLimitMilliseconds]-[dict get ${:data} poolWaitMilliseconds]}]
    if {$remaining<=0} {
        error "aggregation-pool-wait-limit"
    }
    set started [clock milliseconds]
    set :handle [ns_db gethandle -timeout [expr {min(100,$remaining)/1000.0}] [dict get ${:data} pool]]
    # A timed-out ns_db gethandle returns an empty list, not an error.
    # Yield BEFORE parsing or advancing any log bytes; the same slice
    # can therefore be retried without omissions or duplicate counts.
    if {${:handle} eq {}} {
        dict incr :data poolWaitCount
        dict incr :data poolWaitMilliseconds [expr {max(1,[clock milliseconds]-$started)}]
        if {[dict get ${:data} poolWaitMilliseconds]>=[dict get ${:data} poolWaitLimitMilliseconds]} {
            error "aggregation-pool-wait-limit"
        }
        if {[dict get ${:data} status] ne "waiting"} {
            dict set :data retryStatus [dict get ${:data} status]
            dict set :data retryReason [dict get ${:data} reason]
        }
        dict set :data status waiting
        dict set :data reason aggregation-pool-busy
        return
    }
    if {[dict get ${:data} status] eq "waiting"} {
        dict set :data status [:dict getdef ${:data} retryStatus "active"]
        dict set :data reason [:dict getdef ${:data} retryReason {}]
    }
    set phase validate-pool
    if {[ns_db dbtype ${:handle}] ne "sqlite"} {
        error "aggregation-wrong-driver"
    }
    set :db [::nsmcp::DatabaseSession new -handle ${:handle}]
    set phase create-spool
    set create [expr {![dict get ${:data} schemaReady]}]
    if {[dict get ${:data} path] eq {}} {
        set f [file tempfile path [file join ${:directory} access-aggregate-]]
        close $f
        dict set :data path $path
        file attributes $path -permissions 0600
    } else {
        set path [dict get ${:data} path]
    }
    set phase check-spool
    if {![file isfile $path]} {
        error "aggregation-spool-missing"
    }
    set phase detach-spool
    foreach row [${:db} execute {PRAGMA database_list}] {
        if {[dict get $row name] eq "nsmcp_access_aggregate"} {
            ${:db} execute {DETACH DATABASE nsmcp_access_aggregate}
        }
    }
    set phase attach-spool
    ${:db} execute {ATTACH DATABASE $path AS nsmcp_access_aggregate}
    set :attached true
    set phase configure-spool
    ${:db} execute {PRAGMA nsmcp_access_aggregate.journal_mode=DELETE;
        PRAGMA nsmcp_access_aggregate.synchronous=OFF;
        PRAGMA nsmcp_access_aggregate.cache_size=-1024;
        PRAGMA nsmcp_access_aggregate.mmap_size=0;
        PRAGMA nsmcp_access_aggregate.journal_size_limit=0}
    set phase disk-budget
    set pageSize [${:db} scalar {PRAGMA nsmcp_access_aggregate.page_size}]
    set pages [expr {wide([dict get ${:data} maxBytes]*0.4)/$pageSize}]
    if {[${:db} scalar {PRAGMA nsmcp_access_aggregate.page_count}]>$pages} {
        error "aggregation-disk-limit"
    }
    ${:db} execute "PRAGMA nsmcp_access_aggregate.max_page_count=$pages"
    if {$create} {
        set phase create-schema
        set owner [dict get ${:data} owner]
        ${:db} execute {CREATE TABLE nsmcp_access_aggregate.owner (id TEXT PRIMARY KEY);
            INSERT INTO nsmcp_access_aggregate.owner VALUES($owner);
            CREATE TABLE nsmcp_access_aggregate.counts (
                kind INTEGER NOT NULL,k TEXT NOT NULL,label TEXT NOT NULL,
                clipped INTEGER NOT NULL,redacted INTEGER NOT NULL,n INTEGER NOT NULL,
                PRIMARY KEY(kind,k)) WITHOUT ROWID;
            CREATE INDEX nsmcp_access_aggregate.ranking ON counts(kind,n DESC,k)}
        dict set :data schemaReady true
    } else {
        set phase verify-owner
        if {[${:db} scalar {SELECT id FROM nsmcp_access_aggregate.owner}] ne [dict get ${:data} owner]} {
            error "aggregation-owner-mismatch"
        }
    }
} on error {message options} {
    set reason sqlite-initialization-failed
    if {$message in {aggregation-spool-missing aggregation-disk-limit aggregation-pool-wait-limit aggregation-owner-mismatch aggregation-wrong-driver}} {
        set reason $message
    }
    :fail $phase $reason $message $options
    :disconnect true
}

XQL Not present:

Generic, PostgreSQL, Oracle