nsmcp::AccessRecordStoreSQLite method init (protected)

 <instance of nsmcp::AccessRecordStoreSQLite> init

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

Ordered column declarations describe the prepared request rows. Build the core batch schema once, before any ingestion can start.

Testcases:

No testcase defined.

Source code:

set :partitionInterval [:dict getdef ${:settings} partitionInterval 0]
if {![string is integer -strict ${:partitionInterval}] || ${:partitionInterval}<0} {
    error "invalid partition interval"
}
if {${:partitionInterval}>0 && (${:partitionInterval}<3600
    || ${:partitionInterval}%60!=0
    || int(ceil([dict get ${:settings} retention]/double(${:partitionInterval})))+2>256)} {
    error "partition interval must be minute-aligned, at least one hour, and retain at most 256 tables"
}
set columns {}
foreach {name definition} {
    fid        {type number refinement int64 representation integer_decimal}
    off        {type number refinement int64 representation integer_decimal}
    stop       {type number refinement int64 representation integer_decimal}
    kind       {type number refinement int64 representation integer_decimal}
    t          {type number refinement float64 representation decimal}
    status     {type number refinement int64 representation integer_decimal}
    path       {type string}
    method     {type string}
    pool       {type string}
    source     {type string}
    driver     {type string}
    auth       {type string}
    bytes      {type string refinement int64_or_empty_text representation integer_decimal}
    phase      {type number refinement int64 representation integer_decimal}
    fallback   {type number refinement int64 representation integer_decimal}
    a          {type number refinement float64 representation decimal}
    q          {type number refinement float64 representation decimal}
    f          {type number refinement float64 representation decimal}
    r          {type number refinement float64 representation decimal}
    peer       {type string refinement hex_or_empty_text representation hex}
    agent      {type string refinement hex_or_empty_text representation hex}
    label      {type string}
    clipped    {type number refinement int64 representation integer_decimal}
    redacted   {type number refinement int64 representation integer_decimal}
    identifier {type string}
    score      {type string refinement encoded_binary representation hex}
    completed  {type number refinement float64 representation decimal}
    rolled     {type number refinement int64 representation integer_decimal}
} {
    lappend columns [dict merge [dict create name $name] $definition]
}
set :batchSchema [dict create id nsmcp-cooked-access version 2 columns $columns]
if {[dict exists ${:settings} retentionCacheBytes]} {
    set bytes [dict get ${:settings} retentionCacheBytes]
    if {![string is wideinteger -strict $bytes] || $bytes < 1048576 || $bytes > 67108864} {
        error "SQLite retentionCacheBytes must be 1MiB through 64MiB"
    }
    set :retentionCacheKiB [expr {($bytes+1023)/1024}]
}
try {
    set pool [dict get ${:settings} pool]
    set datasource [ns_config ns/db/pool/$pool datasource {}]
    if {$datasource eq {} || [file pathtype $datasource] ne "absolute"} {
        error "accessRecordPool requires a persistent absolute datasource"
    }
    set :path [file normalize $datasource]
    set directory [file dirname ${:path}]
    if {${:mode} ni {reader writer}} {
        error "invalid access store mode"
    }
    if {${:mode} eq "writer" && ![file exists $directory]} {
        file mkdir $directory
        file attributes $directory -permissions 0700
    }
    # Create privately before SQLite opens the file. Never follow a symlink.
    if {${:mode} eq "reader" && ![file exists ${:path}]} {
        error "record-store-not-initialized"
    }
    if {![file exists ${:path}]} {
        if {[catch {
            open ${:path} {WRONLY CREAT EXCL} 0600
        } f createOptions]} {
            if {[lindex [dict get $createOptions -errorcode] 1] ne "EEXIST"} {
                return -options $createOptions $f
            }
        } else {
            close $f
        }
    }
    file lstat ${:path} st
    if {$st(type) ne "file"} {
        error "record-store-invalid-file"
    }
    if {${:mode} eq "writer"} {
        file attributes ${:path} -permissions 0600
    }
    :accessIngestionObserve pool-checkout
    set :handle [ns_db gethandle -timeout 100ms [dict get ${:settings} pool]]
    if {${:handle} eq {}} {
        return
    }
    if {[ns_db dbtype ${:handle}] ne "sqlite"} {
        error "record-store-wrong-driver"
    }
    set :db [::nsmcp::DatabaseSession new -handle ${:handle}]
    if {[file normalize [ns_db datasource ${:handle}]] ne ${:path}} {
        error "record-store-wrong-datasource"
    }
    set :prior [dict create queryOnly [${:db} scalar {PRAGMA query_only}]]
    foreach row [${:db} execute {
        SELECT name FROM main.sqlite_schema
         WHERE type='table' AND name NOT LIKE 'sqlite_%'
           AND NOT EXISTS(SELECT 1 FROM main.sqlite_schema WHERE type='table' AND name='meta')
    }] {
        set name [dict get $row name]
        if {$name ni {meta files scan_counts ranges rows state peer_addresses minutes evictions partitions row_template}
            && ![regexp {^p_[0-9]+$} $name]} {
            error "accessRecordPool must contain only access-record tables"
        }
    }
    set :sqliteSynchronousValue [${:db} scalar {PRAGMA main.synchronous}]
    set :sqliteSynchronous [lindex {OFF NORMAL FULL EXTRA} ${:sqliteSynchronousValue}]
    set :sqliteWalAutoCheckpointPages [${:db} scalar {PRAGMA wal_autocheckpoint}]

    if {${:mode} eq "writer"} {
        file stat ${:path} identity
        set key [ns_crypto::md string -digest sha256 -- [list ${:path} $identity(dev) $identity(ino) [dict get ${:settings} maxBytes]  [expr {[:dict getdef ${:settings} unified false]}]  schema4]]
        if {[file size ${:path}]<512 || ![nsv_exists nsmcp:record-initialized $key]} {
            :accessIngestionObserve initialize-schema
            :initialize
            nsv_set nsmcp:record-initialized $key 1
        }
        # The page budget belongs to this store, not to the pool policy.
        :accessIngestionObserve page-limit
        :pageLimit
    } else {
        if {[${:db} scalar {SELECT version FROM main.meta}] ni {2 3 4}} {
            error "record-store-schema-not-initialized"
        }
    }
    :partitionOpen
    if {${:mode} eq "writer"} {
        # Only migration backfill uses cooked staging. Both ingestion
        # backends insert directly into the canonical table.
        ${:db} execute {
            DROP TABLE IF EXISTS temp.nsmcp_ingest_batch
        }
    } else {
        ${:db} execute {PRAGMA query_only=ON}
    }
} on error {message opts} {
    if {$message eq "record-store-disk-limit"} {
        # The unified writer emits a rate-limited capacity warning.
        set failure {class disk-limit diagnosticId {}}
    } elseif {$message in {record-store-not-initialized record-store-schema-not-initialized record-store-partition-reset-required}} {
        set failure [dict create class [string range $message 13 end] diagnosticId {}]
    } elseif {[regexp -nocase {database.*(locked|busy)} $message]} {
        set failure [dict create class database-locked diagnosticId {}]
    } else {
        set failure [:diagnostic $message $opts]
    }
    dict set opts -errorcode  [list NSMCP RECORD_STORE [dict get $failure class] [dict get $failure diagnosticId]]
    :disconnect
    return -options $opts $message
}

XQL Not present:

Generic, PostgreSQL, Oracle