content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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.
   */
  package org.apache.cassandra.db.virtual;
  
  import java.util.stream.Stream;
  
  import org.apache.cassandra.concurrent.DebuggableTask;
  import org.apache.cassandra.concurrent.SharedExecutorPool;
  import org.apache.cassandra.db.marshal.LongType;
  import org.apache.cassandra.db.marshal.UTF8Type;
  import org.apache.cassandra.dht.LocalPartitioner;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.service.accord.AccordExecutor;
  import org.apache.cassandra.service.accord.AccordService;
  
  import static java.lang.Long.max;
  import static java.util.concurrent.TimeUnit.NANOSECONDS;
  
  import static org.apache.cassandra.utils.Clock.Global.nanoTime;
  
  /**
   * Virtual table that lists currently running queries on the NTR (coordinator) and Read/Mutation (local) stages
   *
   * Example:
   * <pre>
   * cqlsh> SELECT * FROM system_views.queries;
   *
   *  thread_id                   | queued_micros |  running_micros | task
   * ------------------------------+---------------+-----------------+--------------------------------------------------------------------------------
   *  Native-Transport-Requests-7 |         72923 |            7611 |                      QUERY SELECT * FROM system_views.queries; [pageSize = 100]
   *              MutationStage-2 |         18249 |            2084 | Mutation(keyspace='distributed_test_keyspace', key='000000f8', modifications...
   *                  ReadStage-2 |         72447 |           10121 |                                         SELECT * FROM keyspace.table LIMIT 5000
   * </pre>
   */    
  final class QueriesTable extends AbstractVirtualTable
  {
      private static final String TABLE_NAME = "queries";
      private static final String ID = "thread_id";
      private static final String QUEUED = "queued_micros";
      private static final String RUNNING = "running_micros";
      private static final String DESC = "task";
  
      QueriesTable(String keyspace)
      {
          super(TableMetadata.builder(keyspace, TABLE_NAME)
                             .comment("Lists currently running queries")
                             .kind(TableMetadata.Kind.VIRTUAL)
                             .partitioner(new LocalPartitioner(UTF8Type.instance))
                             // The thread name is unique since the id given to each SEPWorker is unique
                             .addPartitionKeyColumn(ID, UTF8Type.instance)
                             .addRegularColumn(QUEUED, LongType.instance)
                             .addRegularColumn(RUNNING, LongType.instance)
                             .addRegularColumn(DESC, UTF8Type.instance)
                             .build());
      }
  
      /**
       * Walks the {@link SharedExecutorPool} workers for any {@link DebuggableTask} instances and populates the table.
       */
      @Override
      public DataSet data()
      {
          SimpleDataSet result = new SimpleDataSet(metadata());
  
          Stream<? extends DebuggableTask.DebuggableTaskRunner> streams = Stream.concat(
              SharedExecutorPool.SHARED.workers(),
              AccordService.started() ? AccordService.instance().executors().stream().flatMap(AccordExecutor::active) : Stream.of()
          );
          streams.forEach(runner -> {
              DebuggableTask task = runner.running();
              if (task == null) return;
  
              String id = runner.id();
              long creationTimeNanos = task.creationTimeNanos();
              long startTimeNanos = task.startTimeNanos();
              long now = nanoTime();
  
              long queuedMicros = NANOSECONDS.toMicros(max((startTimeNanos > 0 ? startTimeNanos : now) - creationTimeNanos, 0));
              long runningMicros = startTimeNanos > 0 ? NANOSECONDS.toMicros(now - startTimeNanos) : 0;
  
              result.row(id)
                    .column(QUEUED, queuedMicros)
                    .column(RUNNING, runningMicros)
                    .column(DESC, task.description());
          });
  
          return result;
      }
  }
  