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  [expr {[dict exists ${:limits} aggregateWaitMs] ? [dict get ${: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 [expr {[dict exists ${:data} retryStatus] ? [dict get ${:data} retryStatus] : "active"}]
        dict set :data reason [expr {[dict exists ${:data} retryReason] ? [dict get ${: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