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.Arrays;
  import java.util.Collection;
  import java.util.Date;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.UUID;
  import java.util.stream.Collectors;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.collect.ImmutableSet;
  
  import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.marshal.TimeUUIDType;
  import org.apache.cassandra.db.marshal.UUIDType;
  import org.apache.cassandra.dht.LocalPartitioner;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.repair.CommonRange;
  import org.apache.cassandra.repair.RepairJobDesc;
  import org.apache.cassandra.repair.state.Completable;
  import org.apache.cassandra.repair.state.CoordinatorState;
  import org.apache.cassandra.repair.state.JobState;
  import org.apache.cassandra.repair.state.ParticipateState;
  import org.apache.cassandra.repair.state.SessionState;
  import org.apache.cassandra.repair.state.State;
  import org.apache.cassandra.repair.state.ValidationState;
  import org.apache.cassandra.schema.Schema;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.service.ActiveRepairService;
  import org.apache.cassandra.utils.TimeUUID;
  
  import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
  
  public class LocalRepairTables
  {
      private LocalRepairTables()
      {
      }
  
      public static Collection<VirtualTable> getAll(String keyspace)
      {
          return Arrays.asList(
          new RepairTable(keyspace),
          new SessionTable(keyspace),
          new JobTable(keyspace),
          new ParticipateTable(keyspace),
          new ValidationTable(keyspace)
          );
      }
  
      private static final String JOB_DESC_COLUMNS = "  repair_id timeuuid,\n" +
                                                     "  session_id timeuuid,\n" +
                                                     "  keyspace_name text,\n" +
                                                     "  table_name text,\n" +
                                                     "  ranges frozen<list<text>>,\n";
  
      static final class RepairTable extends AbstractVirtualTable
      {
          protected RepairTable(String keyspace)
          {
              super(parse(keyspace, "Repair summary",
                          "CREATE TABLE repairs (\n" +
                          stdColumnsWithStatus(true) +
                          "  command_id int,\n" +
                          "  keyspace_name text,\n" +
                          // human readable definition of what the repair is doing
                          "  type text,\n" +
                          // list of all sessions; this is lazy so only once the session is created will it be present, so this dynamically changes within the life of a repair
                          "  sessions frozen<set<timeuuid>>,\n" +
  
                          // options_ maps to RepairOption
                          "  options_parallelism text,\n" +
                          "  options_primary_range boolean,\n" +
                          "  options_incremental boolean,\n" +
                          "  options_trace boolean,\n" +
                          "  options_job_threads int,\n" +
                          "  options_subrange_repair boolean,\n" +
                          "  options_pull_repair boolean,\n" +
                          "  options_force_repair boolean,\n" +
                          "  options_preview_kind text,\n" +
                          "  options_optimise_streams boolean,\n" +
                          "  options_ignore_unreplicated_keyspaces boolean,\n" +
                          "  options_column_families frozen<set<text>>,\n" +
                          "  options_data_centers frozen<set<text>>,\n" +
                          "  options_hosts frozen<set<text>>,\n" +
                          "  options_ranges frozen<set<text>>,\n" +
  
                          "  table_names frozen<list<text>>,\n" +
                          "  ranges frozen<list<list<text>>>,\n" +
                          "  unfiltered_ranges frozen<list<list<text>>>,\n" +
                          "  participants frozen<list<text>>,\n" +
                          stateColumns(CoordinatorState.State.class) +
                          "\n" +
                          "  PRIMARY KEY ( (id) )\n" +
                          ")"));
          }
  
          public DataSet data()
          {
              SimpleDataSet result = new SimpleDataSet(metadata());
              ActiveRepairService.instance().coordinators().forEach(s -> updateDataset(result, s));
              return result;
          }
  
          public DataSet data(DecoratedKey partitionKey)
          {
              TimeUUID id = TimeUUIDType.instance.compose(partitionKey.getKey());
              SimpleDataSet result = new SimpleDataSet(metadata());
              CoordinatorState state = ActiveRepairService.instance().coordinator(id);
              if (state != null)
                  updateDataset(result, state);
              return result;
          }
  
          private void updateDataset(SimpleDataSet result, CoordinatorState state)
          {
              result.row(state.id);
              addState(result, state);
              result.column("type", state.getType());
              result.column("keyspace_name", state.keyspace);
              result.column("command_id", state.cmd);
  
              result.column("options_parallelism", state.options.getParallelism().name());
              result.column("options_primary_range", state.options.isPrimaryRange());
              result.column("options_trace", state.options.isTraced());
              result.column("options_job_threads", state.options.getJobThreads());
              result.column("options_subrange_repair", false);
              result.column("options_pull_repair", state.options.isPullRepair());
              result.column("options_force_repair", state.options.isForcedRepair());
              result.column("options_preview_kind", state.options.getPreviewKind().name());
              result.column("options_optimise_streams", state.options.optimiseStreams());
              result.column("options_ignore_unreplicated_keyspaces", state.options.ignoreUnreplicatedKeyspaces());
              result.column("options_column_families", state.options.getColumnFamilies());
              result.column("options_data_centers", state.options.getDataCenters());
              result.column("options_hosts", state.options.getHosts());
              result.column("options_ranges", toStringSet(state.options.getRanges()));
  
              result.column("sessions", state.getSessionIds());
  
              String[] columnFamilyNames = state.getColumnFamilyNames();
              result.column("table_names", columnFamilyNames == null ? null : Arrays.asList(columnFamilyNames));
  
              Set<InetAddressAndPort> participants = state.getParticipants();
              result.column("participants", participants == null ? null : toStringList(participants));
  
              List<CommonRange> ranges = state.getFilteredCommonRanges();
              result.column("ranges", ranges == null ? null : ranges.stream().map(c -> c.ranges).map(LocalRepairTables::toStringList).collect(Collectors.toList()));
  
              ranges = state.getCommonRanges();
              result.column("unfiltered_ranges", ranges == null ? null : ranges.stream().map(c -> c.ranges).map(LocalRepairTables::toStringList).collect(Collectors.toList()));
          }
      }
  
      private static final class SessionTable extends AbstractVirtualTable
      {
          SessionTable(String keyspace)
          {
              super(parse(keyspace, "Repair session",
                          "CREATE TABLE repair_sessions (\n" +
                          stdColumnsWithStatus(true) +
                          "  repair_id timeuuid,\n" +
                          "  keyspace_name text,\n" +
                          "  table_names frozen<list<text>>,\n" +
                          "  ranges frozen<list<text>>,\n" +
                          "  participants frozen<list<text>>,\n" +
                          "  jobs frozen<set<uuid>>,\n" +
                          "\n" +
                          stateColumns(SessionState.State.class) +
                          "\n" +
                          "  PRIMARY KEY ( (id) )\n" +
                          ")"));
          }
  
          @Override
          public DataSet data()
          {
              SimpleDataSet result = new SimpleDataSet(metadata());
              ActiveRepairService.instance().coordinators().stream()
                                                  .flatMap(s -> s.getSessions().stream())
                                                  .forEach(s -> updateDataset(result, s));
              return result;
          }
  
          private void updateDataset(SimpleDataSet result, SessionState state)
          {
              result.row(state.id);
              addState(result, state);
              result.column("repair_id", state.parentRepairSession);
              result.column("keyspace_name", state.keyspace);
              result.column("table_names", Arrays.asList(state.cfnames));
              result.column("ranges", toStringList(state.commonRange.ranges));
              result.column("jobs", state.getJobIds());
              result.column("participants", toStringList(state.getParticipants()));
          }
      }
  
      private static final class JobTable extends AbstractVirtualTable
      {
          JobTable(String keyspace)
          {
              super(parse(keyspace, "Repair job",
                          "CREATE TABLE repair_jobs (\n" +
                          stdColumnsWithStatus(false) +
                          "  participants frozen<list<text>>,\n" +
                          JOB_DESC_COLUMNS +
                          "\n" +
                          stateColumns(JobState.State.class) +
                          "\n" +
                          "  PRIMARY KEY ( (id) )\n" +
                          ")"));
          }
  
          @Override
          public DataSet data()
          {
              SimpleDataSet result = new SimpleDataSet(metadata());
              ActiveRepairService.instance().coordinators().stream()
                                                  .flatMap(s -> s.getSessions().stream())
                                                  .flatMap(s -> s.getJobs().stream())
                                                  .forEach(s -> updateDataset(result, s));
              return result;
          }
  
          private void updateDataset(SimpleDataSet result, JobState state)
          {
              result.row(state.id);
              addState(result, state);
              addState(result, state.desc);
              result.column("participants", toStringList(state.getParticipants()));
          }
      }
  
      static final class ParticipateTable extends AbstractVirtualTable
      {
          protected ParticipateTable(String keyspace)
          {
              super(parse(keyspace, "Repair participate summary",
                          "CREATE TABLE repair_participates (" +
                          stdColumns(true) +
                          "  initiator  text,\n" +
                          "  tables frozen<set<text>>, \n" +
                          "  ranges frozen<list<text>>,\n" +
                          "  incremental boolean,\n" +
                          "  global boolean,\n" +
                          "  preview_kind text,\n" +
                          "  repaired_at timestamp,\n" +
                          "  validations frozen<set<uuid>>,\n" +
                          "\n" +
                          "  PRIMARY KEY ( (id) )\n" +
                          ")"));
          }
  
          @Override
          public DataSet data()
          {
              SimpleDataSet result = new SimpleDataSet(metadata());
              ActiveRepairService.instance().participates().stream()
                                          .forEach(s -> updateDataset(result, s));
              return result;
          }
  
          @Override
          public DataSet data(DecoratedKey partitionKey)
          {
              TimeUUID id = TimeUUIDType.instance.compose(partitionKey.getKey());
              SimpleDataSet result = new SimpleDataSet(metadata());
              ParticipateState state = ActiveRepairService.instance().participate(id);
              if (state != null)
                  updateDataset(result, state);
              return result;
          }
  
          private void updateDataset(SimpleDataSet result, ParticipateState state)
          {
              result.row(state.id);
              addCompletableState(result, state);
              result.column("initiator", state.initiator.toString());
              result.column("tables", state.tableIds.stream()
                                                    .map(Schema.instance::getTableMetadata)
                                                    .filter(a -> a != null) // getTableMetadata returns null if id isn't know, most likely dropped
                                                    .map(Object::toString)
                                                    .collect(Collectors.toSet()));
              result.column("incremental", state.incremental);
              result.column("global", state.global);
              result.column("preview_kind", state.previewKind.name());
              if (state.repairedAt != 0)
                  result.column("repaired_at", new Date(state.repairedAt));
              result.column("validations", ImmutableSet.copyOf(state.validationIds()));
              result.column("ranges", toStringList(state.ranges));
          }
      }
  
      private static final class ValidationTable extends AbstractVirtualTable
      {
          ValidationTable(String keyspace)
          {
              super(parse(keyspace, "Repair validation",
                          "CREATE TABLE repair_validations (\n" +
                          stdColumnsWithStatus(false) +
                          JOB_DESC_COLUMNS +
                          "  initiator  text,\n" +
                          "  estimated_partitions  bigint,\n" +
                          "  estimated_total_bytes  bigint,\n" +
                          "  partitions_processed  bigint,\n" +
                          "  bytes_read  bigint,\n" +
                          "  progress_percentage float,\n" +
                          "\n" +
                          stateColumns(ValidationState.State.class) +
                          "\n" +
                          "  PRIMARY KEY ( (id) )\n" +
                          ")"));
          }
  
          @Override
          public DataSet data()
          {
              SimpleDataSet result = new SimpleDataSet(metadata());
              ActiveRepairService.instance().validations().stream()
                                                       .forEach(s -> updateDataset(result, s));
              return result;
          }
  
          @Override
          public DataSet data(DecoratedKey partitionKey)
          {
              UUID id = UUIDType.instance.compose(partitionKey.getKey());
              SimpleDataSet result = new SimpleDataSet(metadata());
              ValidationState state = ActiveRepairService.instance().validation(id);
              if (state != null)
                  updateDataset(result, state);
              return result;
          }
  
          private void updateDataset(SimpleDataSet result, ValidationState state)
          {
              result.row(state.id);
              addState(result, state);
              addState(result, state.desc);
              result.column("initiator", state.initiator.toString());
              result.column("estimated_partitions", state.estimatedPartitions == 0 ? null : state.estimatedPartitions);
              result.column("estimated_total_bytes", state.estimatedTotalBytes == 0 ? null : state.estimatedTotalBytes);
              result.column("partitions_processed", state.partitionsProcessed == 0 ? null : state.partitionsProcessed);
              result.column("progress_percentage", round(state.getProgress() * 100));
              result.column("bytes_read", state.bytesRead);
          }
      }
  
      private static String timestampColumnName(Enum<?> e)
      {
          return timestampColumnName(toLowerCaseLocalized(e.name()));
      }
  
      private static String timestampColumnName(String e)
      {
          return "state_" + e + "_timestamp";
      }
  
      private static String stateColumns(Class<? extends Enum<?>> klass)
      {
          StringBuilder sb = new StringBuilder();
          for (Enum<?> e : klass.getEnumConstants())
              sb.append("  ").append(timestampColumnName(e)).append(" timestamp, \n");
          return sb.toString();
      }
  
      private static String stdStateColumns()
      {
          StringBuilder sb = new StringBuilder();
          sb.append("  ").append(timestampColumnName("init")).append(" timestamp, \n");
          for (State.Result.Kind kind : State.Result.Kind.values())
              sb.append("  ").append(timestampColumnName(kind)).append(" timestamp, \n");
          return sb.toString();
      }
  
      private static void addCompletableState(SimpleDataSet ds, Completable<?> state)
      {
          // read timestamp early to see latest data
          ds.column("last_updated_at", new Date(state.getLastUpdatedAtMillis()));
          ds.column("duration_millis", state.getDurationMillis());
          State.Result result = state.getResult();
          ds.column("failure_cause", state.getFailureCause());
          ds.column("success_message", state.getSuccessMessage());
          ds.column(timestampColumnName("init"), new Date(state.getInitializedAtMillis()));
          ds.column("completed", result != null);
  
          if (result != null)
              ds.column(timestampColumnName(result.kind), new Date(state.getLastUpdatedAtMillis()));
      }
  
      private static <T extends Enum<T>> void addState(SimpleDataSet ds, State<T, ?> state)
      {
          addCompletableState(ds, state);
  
          T currentState = state.getStatus();
          State.Result result = state.getResult();
          ds.column("status", result != null ? toLowerCaseLocalized(result.kind.name()) : currentState == null ? "init" : toLowerCaseLocalized(currentState.name()));
          for (Map.Entry<T, Long> e : state.getStateTimesMillis().entrySet())
          {
              if (e.getValue().longValue() != 0)
                  ds.column(timestampColumnName(e.getKey()), new Date(e.getValue()));
          }
      }
  
      @VisibleForTesting
      static float round(float value)
      {
          return Math.round(value * 100.0F) / 100.0F;
      }
  
      private static void addState(SimpleDataSet result, RepairJobDesc desc)
      {
          result.column("repair_id", desc.parentSessionId);
          result.column("session_id", desc.sessionId);
          result.column("keyspace_name", desc.keyspace);
          result.column("table_name", desc.columnFamily);
          result.column("ranges", toStringList(desc.ranges));
      }
  
      private static <T> List<String> toStringList(Collection<T> list)
      {
          if (list == null)
              return null;
          return list.stream().map(Object::toString).collect(Collectors.toList());
      }
  
      private static <T> Set<String> toStringSet(Collection<T> list)
      {
          if (list == null)
              return null;
          return list.stream().map(Object::toString).collect(Collectors.toSet());
      }
  
      private static TableMetadata parse(String keyspace, String comment, String query)
      {
          return CreateTableStatement.parse(query, keyspace)
                                     .comment(comment)
                                     .kind(TableMetadata.Kind.VIRTUAL)
                                     .partitioner(new LocalPartitioner(UUIDType.instance))
                                     .build();
      }
  
      private static String stdColumns(boolean time)
      {
          String str = "  id timeuuid,\n" +
                       "  last_updated_at timestamp,\n" +
                       "  completed boolean,\n" +
                       "  duration_millis bigint,\n" +
                       "  failure_cause text,\n" +
                       "  success_message text,\n";
          if (!time)
              str = str.replace("id timeuuid", "id uuid");
          return str + stdStateColumns();
      }
  
      private static String stdColumnsWithStatus(boolean time)
      {
          return stdColumns(time) + "  status text,\n";
      }
  }
  