src/controller/trace_controller.ts

// Copyright (C) 2018 The Android Open Source Project
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

import { assertExists, assertTrue } from '../base/logging';
import {
  Actions,
  DeferredAction,
} from '../common/actions';
import { cacheTrace } from '../common/cache_manager';
import { TRACE_MARGIN_TIME_S } from '../common/constants';
import { Engine } from '../common/engine';
import { featureFlags, Flag, PERF_SAMPLE_FLAG } from '../common/feature_flags';
import { HttpRpcEngine } from '../common/http_rpc_engine';
import { NUM, NUM_NULL, QueryError, STR, STR_NULL } from '../common/query_result';
import { EngineMode } from '../common/state';
import { TimeSpan, toNs, toNsCeil, toNsFloor } from '../common/time';
import { resetEngineWorker, WasmEngineProxy } from '../common/wasm_engine_proxy';
import {
  globals as frontendGlobals,
  QuantizedLoad,
  ThreadDesc
} from '../frontend/globals';
import {
  publishHasFtrace,
  publishMetricError,
  publishOverviewData,
  publishThreads
} from '../frontend/publish';
import { Router } from '../frontend/router';

import {
  CounterAggregationController
} from './aggregation/counter_aggregation_controller';
import {
  CpuAggregationController
} from './aggregation/cpu_aggregation_controller';
import {
  CpuByProcessAggregationController
} from './aggregation/cpu_by_process_aggregation_controller';
import {
  FrameAggregationController
} from './aggregation/frame_aggregation_controller';
import {
  SliceAggregationController
} from './aggregation/slice_aggregation_controller';
import {
  ThreadAggregationController
} from './aggregation/thread_aggregation_controller';
import { Child, Children, Controller } from './controller';
import {
  CpuProfileController,
  CpuProfileControllerArgs
} from './cpu_profile_controller';
import {
  FlamegraphController,
  FlamegraphControllerArgs
} from './flamegraph_controller';
import {
  FlowEventsController,
  FlowEventsControllerArgs
} from './flow_events_controller';
import { globals } from './globals';
import { LoadingManager } from './loading_manager';
import { LogsController } from './logs_controller';
import { MetricsController } from './metrics_controller';
import {
  PivotTableController,
  PivotTableControllerArgs
} from './pivot_table_controller';
import {PivotTableReduxController} from './pivot_table_redux_controller';
import {QueryController, QueryControllerArgs} from './query_controller';
import {SearchController} from './search_controller';
import {
  SelectionController,
  SelectionControllerArgs
} from './selection_controller';
import {
  TraceErrorController,
} from './trace_error_controller';
import {
  TraceBufferStream,
  TraceFileStream,
  TraceHttpStream,
  TraceStream
} from './trace_stream';
import { TrackControllerArgs, trackControllerRegistry } from './track_controller';
import { decideTracks } from './track_decider';

type States = 'init' | 'loading_trace' | 'ready';

const METRICS = [
  'android_startup',
  'android_ion',
  'android_lmk',
  'android_dma_heap',
  'android_thread_time_in_state',
  'android_surfaceflinger',
  'android_batt',
  'android_sysui_cuj',
  'android_jank',
  'android_camera',
  'android_other_traces',
  'chrome_dropped_frames',
  'chrome_long_latency',
  'trace_metadata',
  'android_trusty_workqueues',
];
const FLAGGED_METRICS: Array<[Flag, string]> = METRICS.map(m => {
  const id = `forceMetric${m}`;
  let name = m.split('_').join(' ') + ' metric';
  name = name[0].toUpperCase() + name.slice(1);
  const flag = featureFlags.register({
    id,
    name,
    description: `Overrides running the '${m}' metric at import time.`,
    defaultValue: true,
  });
  return [flag, m];
});

// TraceController handles handshakes with the frontend for everything that
// concerns a single trace. It owns the WASM trace processor engine, handles
// tracks data and SQL queries. There is one TraceController instance for each
// trace opened in the UI (for now only one trace is supported).
export class TraceController extends Controller<States> {
  private readonly engineId: string;
  private engine?: Engine;

  constructor(engineId: string) {
    super('init');
    this.engineId = engineId;
  }

  run() {
    const engineCfg = assertExists(globals.state.engines[this.engineId]);
    switch (this.state) {
      case 'init':
        this.loadTrace()
          .then(mode => {
            globals.dispatch(Actions.setEngineReady({
              engineId: this.engineId,
              ready: true,
              mode,
            }));
          })
          .catch(err => {
            this.updateStatus(`${err}`);
            throw err;
          });
        this.updateStatus('Opening trace');
        this.setState('loading_trace');
        break;

      case 'loading_trace':
        // Stay in this state until loadTrace() returns and marks the engine as
        // ready.
        if (this.engine === undefined || !engineCfg.ready) return;
        this.setState('ready');
        break;

      case 'ready':
        // At this point we are ready to serve queries and handle tracks.
        const engine = assertExists(this.engine);
        const childControllers: Children = [];

        // Create a TrackController for each track.
        for (const trackId of Object.keys(globals.state.tracks)) {
          const trackCfg = globals.state.tracks[trackId];
          if (trackCfg.engineId !== this.engineId) continue;
          if (!trackControllerRegistry.has(trackCfg.kind)) continue;
          const trackCtlFactory = trackControllerRegistry.get(trackCfg.kind);
          const trackArgs: TrackControllerArgs = { trackId, engine };
          childControllers.push(Child(trackId, trackCtlFactory, trackArgs));
        }

        // Create a QueryController for each query.
        for (const queryId of Object.keys(globals.state.queries)) {
          const queryArgs: QueryControllerArgs = { queryId, engine };
          childControllers.push(Child(queryId, QueryController, queryArgs));
        }

        const selectionArgs: SelectionControllerArgs = { engine };
        childControllers.push(
          Child('selection', SelectionController, selectionArgs));

        const flowEventsArgs: FlowEventsControllerArgs = { engine };
        childControllers.push(
          Child('flowEvents', FlowEventsController, flowEventsArgs));

        const cpuProfileArgs: CpuProfileControllerArgs = { engine };
        childControllers.push(
          Child('cpuProfile', CpuProfileController, cpuProfileArgs));

        const flamegraphArgs: FlamegraphControllerArgs = { engine };
        childControllers.push(
          Child('flamegraph', FlamegraphController, flamegraphArgs));
        childControllers.push(Child(
          'cpu_aggregation',
          CpuAggregationController,
          { engine, kind: 'cpu_aggregation' }));
        childControllers.push(Child(
          'thread_aggregation',
          ThreadAggregationController,
          { engine, kind: 'thread_state_aggregation' }));
        childControllers.push(Child(
          'cpu_process_aggregation',
          CpuByProcessAggregationController,
          { engine, kind: 'cpu_by_process_aggregation' }));
        childControllers.push(Child(
          'slice_aggregation',
          SliceAggregationController,
          { engine, kind: 'slice_aggregation' }));
        childControllers.push(Child(
          'counter_aggregation',
          CounterAggregationController,
          { engine, kind: 'counter_aggregation' }));
        childControllers.push(Child(
          'frame_aggregation',
          FrameAggregationController,
          { engine, kind: 'frame_aggregation' }));
        childControllers.push(Child('search', SearchController, {
          engine,
          app: globals,
        }));
        childControllers.push(
            Child('pivot_table_redux', PivotTableReduxController, {engine}));

        childControllers.push(Child('logs', LogsController, {
          engine,
          app: globals,
        }));
        childControllers.push(
          Child('traceError', TraceErrorController, { engine }));
        childControllers.push(Child('metrics', MetricsController, { engine }));

        // Create a PivotTableController for each pivot table.
        for (const pivotTableId of Object.keys(globals.state.pivotTable)) {
          const pivotTableArgs:
            PivotTableControllerArgs = { pivotTableId, engine };
          childControllers.push(
            Child(pivotTableId, PivotTableController, pivotTableArgs));
        }

        return childControllers;

      default:
        throw new Error(`unknown state ${this.state}`);
    }
    return;
  }

  onDestroy() {
    frontendGlobals.engines.delete(this.engineId);
  }

  private async loadTrace(): Promise<EngineMode> {
    this.updateStatus('Creating trace processor');
    // Check if there is any instance of the trace_processor_shell running in
    // HTTP RPC mode (i.e. trace_processor_shell -D).
    let engineMode: EngineMode;
    let useRpc = false;
    if (globals.state.newEngineMode === 'USE_HTTP_RPC_IF_AVAILABLE') {
      useRpc = (await HttpRpcEngine.checkConnection()).connected;
    }
    let engine;
    if (useRpc) {
      console.log('Opening trace using native accelerator over HTTP+RPC');
      engineMode = 'HTTP_RPC';
      engine = new HttpRpcEngine(this.engineId, LoadingManager.getInstance);
      engine.errorHandler = (err) => {
        globals.dispatch(
          Actions.setEngineFailed({ mode: 'HTTP_RPC', failure: `${err}` }));
        throw err;
      };
    } else {
      console.log('Opening trace using built-in WASM engine');
      engineMode = 'WASM';
      const enginePort = resetEngineWorker();
      engine = new WasmEngineProxy(
        this.engineId, enginePort, LoadingManager.getInstance);
    }
    this.engine = engine;

    frontendGlobals.engines.set(this.engineId, engine);
    globals.dispatch(Actions.setEngineReady({
      engineId: this.engineId,
      ready: false,
      mode: engineMode,
    }));
    const engineCfg = assertExists(globals.state.engines[this.engineId]);
    let traceStream: TraceStream | undefined;
    if (engineCfg.source.type === 'FILE') {
      traceStream = new TraceFileStream(engineCfg.source.file);
    } else if (engineCfg.source.type === 'ARRAY_BUFFER') {
      traceStream = new TraceBufferStream(engineCfg.source.buffer);
    } else if (engineCfg.source.type === 'URL') {
      traceStream = new TraceHttpStream(engineCfg.source.url);
    } else if (engineCfg.source.type === 'HTTP_RPC') {
      traceStream = undefined;
    } else {
      throw new Error(`Unknown source: ${JSON.stringify(engineCfg.source)}`);
    }

    // |traceStream| can be undefined in the case when we are using the external
    // HTTP+RPC endpoint and the trace processor instance has already loaded
    // a trace (because it was passed as a cmdline argument to
    // trace_processor_shell). In this case we don't want the UI to load any
    // file/stream and we just want to jump to the loading phase.
    if (traceStream !== undefined) {
      const tStart = performance.now();
      for (; ;) {
        const res = await traceStream.readChunk();
        await this.engine.parse(res.data);
        const elapsed = (performance.now() - tStart) / 1000;
        let status = 'Loading trace ';
        if (res.bytesTotal > 0) {
          const progress = Math.round(res.bytesRead / res.bytesTotal * 100);
          status += `${progress}%`;
        } else {
          status += `${Math.round(res.bytesRead / 1e6)} MB`;
        }
        status += ` - ${Math.ceil(res.bytesRead / elapsed / 1e6)} MB/s`;
        this.updateStatus(status);
        if (res.eof) break;
      }
      await this.engine.notifyEof();
    } else {
      assertTrue(this.engine instanceof HttpRpcEngine);
      await this.engine.restoreInitialTables();
    }

    // traceUuid will be '' if the trace is not cacheable (URL or RPC).
    const traceUuid = await this.cacheCurrentTrace();

    const traceTime = await this.engine.getTraceTimeBounds();
    let startSec = traceTime.start;
    let endSec = traceTime.end;
    startSec -= TRACE_MARGIN_TIME_S;
    endSec += TRACE_MARGIN_TIME_S;
    const traceTimeState = {
      startSec,
      endSec,
    };

    const emptyOmniboxState = {
      omnibox: '',
      mode: frontendGlobals.state.frontendLocalState.omniboxState.mode ||
        'SEARCH',
      lastUpdate: Date.now() / 1000
    };

    const actions: DeferredAction[] = [
      Actions.setOmnibox(emptyOmniboxState),
      Actions.setTraceUuid({ traceUuid }),
      Actions.setTraceTime(traceTimeState)
    ];

    let visibleStartSec = startSec;
    let visibleEndSec = endSec;
    const mdTime = await this.engine.getTracingMetadataTimeBounds();
    // make sure the bounds hold
    if (Math.max(visibleStartSec, mdTime.start - TRACE_MARGIN_TIME_S) <
      Math.min(visibleEndSec, mdTime.end + TRACE_MARGIN_TIME_S)) {
      visibleStartSec =
        Math.max(visibleStartSec, mdTime.start - TRACE_MARGIN_TIME_S);
      visibleEndSec = Math.min(visibleEndSec, mdTime.end + TRACE_MARGIN_TIME_S);
    }

    // We don't know the resolution at this point. However this will be
    // replaced in 50ms so a guess is fine.
    const resolution = (visibleStartSec - visibleEndSec) / 1000;
    actions.push(Actions.setVisibleTraceTime({
      startSec: visibleStartSec,
      endSec: visibleEndSec,
      lastUpdate: Date.now() / 1000,
      resolution
    }));

    globals.dispatchMultiple(actions);
    Router.navigate(`#!/viewer?local_cache_key=${traceUuid}`);

    // Make sure the helper views are available before we start adding tracks.
    await this.initialiseHelperViews();

    {
      // When we reload from a permalink don't create extra tracks:
      const { pinnedTracks, tracks } = globals.state;
      if (!pinnedTracks.length && !Object.keys(tracks).length) {
        await this.listTracks();
      }
    }

    await this.listThreads();
    await this.loadTimelineOverview(traceTime);

    {
      // A quick heuristic to check if the trace has ftrace events. This is
      // based on the assumption that most traces that have ftrace either:
      // - Are proto traces captured via perfetto, in which case traced_probes
      //   emits ftrace per-cpu stats that end up in the stats table.
      // - Have a raw event with non-zero cpu or utid.
      // Notes:
      // - The "+1 > 1" is to avoid pushing down the constraints to the "raw"
      //   table, which would compute a full column filter without being aware
      //   of the limit 1, and instead delegate the filtering to the iterator.
      const query = `select '_' as _ from raw
          where cpu + 1 > 1 or utid + 1 > 1 limit 1`;
      const result = await assertExists(this.engine).query(query);
      const hasFtrace = result.numRows() > 0;
      publishHasFtrace(hasFtrace);
    }

    globals.dispatch(Actions.removeDebugTrack({}));
    globals.dispatch(Actions.sortThreadTracks({}));

    await this.selectFirstHeapProfile();
    if (PERF_SAMPLE_FLAG.get()) {
      await this.selectPerfSample();
    }

    return engineMode;
  }

  private async selectPerfSample() {
    const query = `select ts, upid
        from perf_sample
        join thread using (utid)
        order by ts desc limit 1`;
    const profile = await assertExists(this.engine).query(query);
    if (profile.numRows() !== 1) return;
    const row = profile.firstRow({ ts: NUM, upid: NUM });
    const ts = row.ts;
    const upid = row.upid;
    globals.dispatch(
      Actions.selectPerfSamples({ id: 0, upid, ts, type: 'perf' }));
  }

  private async selectFirstHeapProfile() {
    const query = `select * from
    (select distinct(ts) as ts, 'native' as type,
        upid from heap_profile_allocation
        union
        select distinct(graph_sample_ts) as ts, 'graph' as type, upid from
        heap_graph_object) order by ts limit 1`;
    const profile = await assertExists(this.engine).query(query);
    if (profile.numRows() !== 1) return;
    const row = profile.firstRow({ ts: NUM, type: STR, upid: NUM });
    const ts = row.ts;
    const type = row.type;
    const upid = row.upid;
    globals.dispatch(Actions.selectHeapProfile({ id: 0, upid, ts, type }));
  }

  private async listTracks() {
    this.updateStatus('Loading tracks');
    const engine = assertExists<Engine>(this.engine);
    const actions = await decideTracks(this.engineId, engine);
    globals.dispatchMultiple(actions);
  }

  private async listThreads() {
    this.updateStatus('Reading thread list');
    const query = `select
        utid,
        tid,
        pid,
        ifnull(thread.name, '') as threadName,
        ifnull(
          case when length(process.name) > 0 then process.name else null end,
          thread.name) as procName,
        process.cmdline as cmdline
        from (select * from thread order by upid) as thread
        left join (select * from process order by upid) as process
        using(upid)`;
    const result = await assertExists(this.engine).query(query);
    const threads: ThreadDesc[] = [];
    const it = result.iter({
      utid: NUM,
      tid: NUM,
      pid: NUM_NULL,
      threadName: STR,
      procName: STR_NULL,
      cmdline: STR_NULL,
    });
    for (; it.valid(); it.next()) {
      const utid = it.utid;
      const tid = it.tid;
      const pid = it.pid === null ? undefined : it.pid;
      const threadName = it.threadName;
      const procName = it.procName === null ? undefined : it.procName;
      const cmdline = it.cmdline === null ? undefined : it.cmdline;
      threads.push({ utid, tid, threadName, pid, procName, cmdline });
    }
    publishThreads(threads);
  }

  private async loadTimelineOverview(traceTime: TimeSpan) {
    const engine = assertExists<Engine>(this.engine);
    const numSteps = 100;
    const stepSec = traceTime.duration / numSteps;
    let hasSchedOverview = false;
    for (let step = 0; step < numSteps; step++) {
      this.updateStatus(
        'Loading overview ' +
        `${Math.round((step + 1) / numSteps * 1000) / 10}%`);
      const startSec = traceTime.start + step * stepSec;
      const startNs = toNsFloor(startSec);
      const endSec = startSec + stepSec;
      const endNs = toNsCeil(endSec);

      // Sched overview.
      const schedResult = await engine.query(
        `select sum(dur)/${stepSec}/1e9 as load, cpu from sched ` +
        `where ts >= ${startNs} and ts < ${endNs} and utid != 0 ` +
        'group by cpu order by cpu');
      const schedData: { [key: string]: QuantizedLoad } = {};
      const it = schedResult.iter({ load: NUM, cpu: NUM });
      for (; it.valid(); it.next()) {
        const load = it.load;
        const cpu = it.cpu;
        schedData[cpu] = { startSec, endSec, load };
        hasSchedOverview = true;
      }
      publishOverviewData(schedData);
    }

    if (hasSchedOverview) {
      return;
    }

    // Slices overview.
    const traceStartNs = toNs(traceTime.start);
    const stepSecNs = toNs(stepSec);
    const sliceResult = await engine.query(`select
           bucket,
           upid,
           ifnull(sum(utid_sum) / cast(${stepSecNs} as float), 0) as load
         from thread
         inner join (
           select
             ifnull(cast((ts - ${traceStartNs})/${stepSecNs} as int), 0) as bucket,
             sum(dur) as utid_sum,
             utid
           from slice
           inner join thread_track on slice.track_id = thread_track.id
           group by bucket, utid
         ) using(utid)
         where upid is not null
         group by bucket, upid`);

    const slicesData: { [key: string]: QuantizedLoad[] } = {};
    const it = sliceResult.iter({ bucket: NUM, upid: NUM, load: NUM });
    for (; it.valid(); it.next()) {
      const bucket = it.bucket;
      const upid = it.upid;
      const load = it.load;

      const startSec = traceTime.start + stepSec * bucket;
      const endSec = startSec + stepSec;

      const upidStr = upid.toString();
      let loadArray = slicesData[upidStr];
      if (loadArray === undefined) {
        loadArray = slicesData[upidStr] = [];
      }
      loadArray.push({ startSec, endSec, load });
    }
    publishOverviewData(slicesData);
  }

  private async cacheCurrentTrace(): Promise<string> {
    const engine = assertExists(this.engine);
    const result = await engine.query(`select str_value as uuid from metadata
                  where name = 'trace_uuid'`);
    if (result.numRows() === 0) {
      // One of the cases covered is an empty trace.
      return '';
    }
    const traceUuid = result.firstRow({ uuid: STR }).uuid;
    const engineConfig = assertExists(globals.state.engines[engine.id]);
    if (!(await cacheTrace(engineConfig.source, traceUuid))) {
      // If the trace is not cacheable (cacheable means it has been opened from
      // URL or RPC) only append '?local_cache_key' to the URL, without the
      // local_cache_key value. Doing otherwise would cause an error if the tab
      // is discarded or the user hits the reload button because the trace is
      // not in the cache.
      return '';
    }
    return traceUuid;
  }

  async initialiseHelperViews() {
    const engine = assertExists<Engine>(this.engine);

    this.updateStatus('Creating annotation counter track table');
    // Create the helper tables for all the annotations related data.
    // NULL in min/max means "figure it out per track in the usual way".
    await engine.query(`
      CREATE TABLE annotation_counter_track(
        id INTEGER PRIMARY KEY,
        name STRING,
        __metric_name STRING,
        upid INTEGER,
        min_value DOUBLE,
        max_value DOUBLE
      );
    `);
    this.updateStatus('Creating annotation slice track table');
    await engine.query(`
      CREATE TABLE annotation_slice_track(
        id INTEGER PRIMARY KEY,
        name STRING,
        __metric_name STRING,
        upid INTEGER,
        group_name STRING
      );
    `);

    this.updateStatus('Creating annotation counter table');
    await engine.query(`
      CREATE TABLE annotation_counter(
        id BIG INT,
        track_id INT,
        ts BIG INT,
        value DOUBLE,
        PRIMARY KEY (track_id, ts)
      ) WITHOUT ROWID;
    `);
    this.updateStatus('Creating annotation slice table');
    await engine.query(`
      CREATE TABLE annotation_slice(
        id INTEGER PRIMARY KEY,
        track_id INT,
        ts BIG INT,
        dur BIG INT,
        depth INT,
        cat STRING,
        name STRING,
        UNIQUE(track_id, ts)
      );
    `);


    const availableMetrics = [];
    const metricsResult = await engine.query('select name from trace_metrics');
    for (const it = metricsResult.iter({name: STR}); it.valid(); it.next()) {
      availableMetrics.push(it.name);
    }
    globals.dispatch(Actions.setAvailableMetrics({availableMetrics}));

    const availableMetricsSet = new Set<string>(availableMetrics);
    for (const [flag, metric] of FLAGGED_METRICS) {
      if (!flag.get() || !availableMetricsSet.has(metric)) {
        continue;
      }

      this.updateStatus(`Computing ${metric} metric`);
      try {
        // We don't care about the actual result of metric here as we are just
        // interested in the annotation tracks.
        await engine.computeMetric([metric]);
      } catch (e) {
        if (e instanceof QueryError) {
          publishMetricError('MetricError: ' + e.message);
          continue;
        } else {
          throw e;
        }
      }

      this.updateStatus(`Inserting data for ${metric} metric`);
      try {
        const result = await engine.query(`pragma table_info(${metric}_event)`);
        let hasSliceName = false;
        let hasDur = false;
        let hasUpid = false;
        let hasValue = false;
        let hasGroupName = false;
        const it = result.iter({ name: STR });
        for (; it.valid(); it.next()) {
          const name = it.name;
          hasSliceName = hasSliceName || name === 'slice_name';
          hasDur = hasDur || name === 'dur';
          hasUpid = hasUpid || name === 'upid';
          hasValue = hasValue || name === 'value';
          hasGroupName = hasGroupName || name === 'group_name';
        }

        const upidColumnSelect = hasUpid ? 'upid' : '0 AS upid';
        const upidColumnWhere = hasUpid ? 'upid' : '0';
        const groupNameColumn =
          hasGroupName ? 'group_name' : 'NULL AS group_name';
        if (hasSliceName && hasDur) {
          await engine.query(`
            INSERT INTO annotation_slice_track(
              name, __metric_name, upid, group_name)
            SELECT DISTINCT
              track_name,
              '${metric}' as metric_name,
              ${upidColumnSelect},
              ${groupNameColumn}
            FROM ${metric}_event
            WHERE track_type = 'slice'
          `);
          await engine.query(`
            INSERT INTO annotation_slice(track_id, ts, dur, depth, cat, name)
            SELECT
              t.id AS track_id,
              ts,
              dur,
              0 AS depth,
              a.track_name as cat,
              slice_name AS name
            FROM ${metric}_event a
            JOIN annotation_slice_track t
            ON a.track_name = t.name AND t.__metric_name = '${metric}'
            ORDER BY t.id, ts
          `);
        }

        if (hasValue) {
          const minMax = await engine.query(`
            SELECT
              IFNULL(MIN(value), 0) as minValue,
              IFNULL(MAX(value), 0) as maxValue
            FROM ${metric}_event
            WHERE ${upidColumnWhere} != 0`);
          const row = minMax.firstRow({ minValue: NUM, maxValue: NUM });
          await engine.query(`
            INSERT INTO annotation_counter_track(
              name, __metric_name, min_value, max_value, upid)
            SELECT DISTINCT
              track_name,
              '${metric}' as metric_name,
              CASE ${upidColumnWhere} WHEN 0 THEN NULL ELSE ${row.minValue} END,
              CASE ${upidColumnWhere} WHEN 0 THEN NULL ELSE ${row.maxValue} END,
              ${upidColumnSelect}
            FROM ${metric}_event
            WHERE track_type = 'counter'
          `);
          await engine.query(`
            INSERT INTO annotation_counter(id, track_id, ts, value)
            SELECT
              -1 as id,
              t.id AS track_id,
              ts,
              value
            FROM ${metric}_event a
            JOIN annotation_counter_track t
            ON a.track_name = t.name AND t.__metric_name = '${metric}'
            ORDER BY t.id, ts
          `);
        }
      } catch (e) {
        if (e instanceof QueryError) {
          publishMetricError('MetricError: ' + e.message);
        } else {
          throw e;
        }
      }
    }
  }

  private updateStatus(msg: string): void {
    globals.dispatch(Actions.updateStatus({
      msg,
      timestamp: Date.now() / 1000,
    }));
  }
}