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.tools.nodetool;
  
  import java.io.PrintStream;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import javax.management.openmbean.CompositeData;
  
  import com.google.common.base.Joiner;
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Lists;
  import org.apache.commons.lang3.StringUtils;
  
  import org.apache.cassandra.repair.consistent.LocalSessionInfo;
  import org.apache.cassandra.repair.consistent.admin.CleanupSummary;
  import org.apache.cassandra.repair.consistent.admin.PendingStats;
  import org.apache.cassandra.repair.consistent.admin.RepairStats;
  import org.apache.cassandra.tools.NodeProbe;
  import org.apache.cassandra.tools.nodetool.layout.CassandraUsage;
  import org.apache.cassandra.utils.FBUtilities;
  import picocli.CommandLine.Command;
  import picocli.CommandLine.Option;
  import picocli.CommandLine.Parameters;
  
  import static org.apache.cassandra.tools.nodetool.CommandUtils.concatArgs;
  
  /**
   * Supports listing and failing incremental repair sessions
   */
  @Command(name = "repair_admin",
           description = "list and fail incremental repair sessions",
           subcommands = {RepairAdmin.ListCmd.class,
                          RepairAdmin.SummarizePendingCmd.class,
                          RepairAdmin.SummarizeRepairedCmd.class,
                          RepairAdmin.CleanupDataCmd.class,
                          RepairAdmin.CancelCmd.class})
  public class RepairAdmin extends AbstractCommand
  {
      @Override
      protected void execute(NodeProbe probe)
      {
          AbstractCommand cmd = new ListCmd();
          cmd.probe(probe);
          cmd.logger(output);
          cmd.run();
      }
  
      @Command(name = "list", description = "list repair sessions")
      public static class ListCmd extends AbstractCommand
      {
          @Option(paramLabel = "all", names = { "-a", "--all" }, description = "include completed and failed sessions")
          private boolean all;
  
          @Option(paramLabel = "start_token", names = { "-st", "--start-token" }, description = "Use -st to specify a token at which the repair range starts")
          private String startToken = StringUtils.EMPTY;
  
          @Option(paramLabel = "end_token", names = { "-et", "--end-token" }, description = "Use -et to specify a token at which repair range ends")
          private String endToken = StringUtils.EMPTY;
  
          protected void execute(NodeProbe probe)
          {
              PrintStream out = probe.output().out;
              List<Map<String, String>> sessions = probe.getRepairServiceProxy().getSessions(all, getRangeString(startToken, endToken));
              if (sessions.isEmpty())
              {
                  out.println("no sessions");
              }
              else
              {
                  List<List<String>> rows = new ArrayList<>();
                  rows.add(Lists.newArrayList("id",
                                              "state",
                                              "last activity",
                                              "coordinator",
                                              "participants",
                                              "participants_wp"));
                  long now = FBUtilities.nowInSeconds();
                  for (Map<String, String> session : sessions)
                  {
                      int updated = Integer.parseInt(session.get(LocalSessionInfo.LAST_UPDATE));
                      List<String> values = Lists.newArrayList(session.get(LocalSessionInfo.SESSION_ID),
                                                               session.get(LocalSessionInfo.STATE),
                                                               (now - updated) + " (s)",
                                                               session.get(LocalSessionInfo.COORDINATOR),
                                                               session.get(LocalSessionInfo.PARTICIPANTS),
                                                               session.get(LocalSessionInfo.PARTICIPANTS_WP));
                      rows.add(values);
                  }
  
                  printTable(rows, out);
              }
          }
      }
      @Command(name = "summarize-pending", description = "report the amount of data marked pending repair for the given token " +
                                                         "range (or all replicated range if no tokens are provided")
      public static class SummarizePendingCmd extends AbstractCommand
      {
          @Option(paramLabel = "verbose", names = { "-v", "--verbose" }, description = "print additional info ")
          private boolean verbose;
  
          @Option(paramLabel = "start_token", names = { "-st", "--start-token" }, description = "Use -st to specify a token at which the repair range starts")
          private String startToken = StringUtils.EMPTY;
  
          @Option(paramLabel = "end_token", names = { "-et", "--end-token" }, description = "Use -et to specify a token at which repair range ends")
          private String endToken = StringUtils.EMPTY;
  
          @CassandraUsage(usage = "[<keyspace> <tables>...]", description = "The keyspace followed by one or many tables")
          private List<String> schemaArgs = new ArrayList<>();
  
          @Parameters(index = "0", description = "The keyspace followed by one or many tables", arity = "0..1")
          private String keyspace;
  
          @Parameters(index = "1..*", description = "The tables", arity = "0..*")
          private List<String> tables;
  
          protected void execute(NodeProbe probe)
          {
              schemaArgs = concatArgs(keyspace, tables);
              List<CompositeData> cds = probe.getRepairServiceProxy().getPendingStats(schemaArgs, getRangeString(startToken, endToken));
              List<PendingStats> stats = new ArrayList<>(cds.size());
              cds.forEach(cd -> stats.add(PendingStats.fromComposite(cd)));
  
              stats.sort((l, r) -> {
                  int cmp = l.keyspace.compareTo(r.keyspace);
                  if (cmp != 0)
                      return cmp;
  
                  return l.table.compareTo(r.table);
              });
  
              List<String> header = Lists.newArrayList("keyspace", "table", "total");
              if (verbose)
              {
                  header.addAll(Lists.newArrayList("pending", "finalized", "failed"));
              }
  
              List<List<String>> rows = new ArrayList<>(stats.size() + 1);
              rows.add(header);
  
              for (PendingStats stat : stats)
              {
                  List<String> row = new ArrayList<>(header.size());
  
                  row.add(stat.keyspace);
                  row.add(stat.table);
                  row.add(stat.total.sizeString());
                  if (verbose)
                  {
                      row.add(stat.pending.sizeString());
                      row.add(stat.finalized.sizeString());
                      row.add(stat.failed.sizeString());
                  }
                  rows.add(row);
              }
  
              printTable(rows, probe.output().out);
          }
      }
  
      @Command(name = "summarize-repaired", description = "return the most recent repairedAt timestamp for the given token range " +
                                                          "(or all replicated ranges if no tokens are provided)")
      public static class SummarizeRepairedCmd extends AbstractCommand
      {
          @Option(paramLabel = "verbose", names = {"-v", "--verbose"}, description = "print additional info ")
          private boolean verbose = false;
  
          @Option(paramLabel = "start_token", names = {"-st", "--start-token"}, description = "Use -st to specify a token at which the repair range starts")
          private String startToken = StringUtils.EMPTY;
  
          @Option(paramLabel = "end_token", names = {"-et", "--end-token"}, description = "Use -et to specify a token at which repair range ends")
          private String endToken = StringUtils.EMPTY;
  
          @CassandraUsage(usage = "[<keyspace> <tables>...]", description = "The keyspace followed by one or many tables")
          private List<String> schemaArgs = new ArrayList<>();
  
          @Parameters(index = "0", description = "The keyspace followed by one or many tables", arity = "0..1")
          private String keyspace;
  
          @Parameters(index = "1..*", description = "The tables", arity = "0..*")
          private List<String> tables;
  
          protected void execute(NodeProbe probe)
          {
              schemaArgs = concatArgs(keyspace, tables);
              PrintStream out = probe.output().out;
              List<CompositeData> compositeData = probe.getRepairServiceProxy().getRepairStats(schemaArgs, getRangeString(startToken, endToken));
  
              if (compositeData.isEmpty())
              {
                  out.println("no stats");
                  return;
              }
  
              List<RepairStats> stats = new ArrayList<>(compositeData.size());
              compositeData.forEach(cd -> stats.add(RepairStats.fromComposite(cd)));
  
              stats.sort((l, r) -> {
                  int cmp = l.keyspace.compareTo(r.keyspace);
                  if (cmp != 0)
                      return cmp;
  
                  return l.table.compareTo(r.table);
              });
  
              List<String> header = Lists.newArrayList("keyspace", "table", "min_repaired", "max_repaired");
              if (verbose)
                  header.add("detail");
  
              List<List<String>> rows = new ArrayList<>(stats.size() + 1);
              rows.add(header);
  
              for (RepairStats stat : stats)
              {
                  List<String> row = Lists.newArrayList(stat.keyspace,
                                                        stat.table,
                                                        Long.toString(stat.minRepaired),
                                                        Long.toString(stat.maxRepaired));
                  if (verbose)
                  {
                      row.add(Joiner.on(", ").join(Iterables.transform(stat.sections, RepairStats.Section::toString)));
                  }
                  rows.add(row);
              }
  
              printTable(rows, out);
          }
      }
  
      @Command(name = "cleanup", description = "cleans up pending data from completed sessions. " +
                                               "This happens automatically, but the command is provided " +
                                               "for situations where it needs to be expedited." +
                                              " Use --force to cancel compactions that are preventing promotion")
      public static class CleanupDataCmd extends AbstractCommand
      {
          @Option(paramLabel = "force", names = { "-f", "--force" }, description = "Force a cleanup.")
          private boolean force = false;
  
          @Option(paramLabel = "start_token", names = { "-st", "--start-token" }, description = "Use -st to specify a token at which the repair range starts")
          private String startToken = StringUtils.EMPTY;
  
          @Option(paramLabel = "end_token", names = { "-et", "--end-token" }, description = "Use -et to specify a token at which repair range ends")
          private String endToken = StringUtils.EMPTY;
  
          @CassandraUsage(usage = "[<keyspace> <tables>...]", description = "The keyspace followed by one or many tables")
          private List<String> schemaArgs = new ArrayList<>();
  
          @Parameters(index = "0", description = "The keyspace followed by one or many tables", arity = "0..1")
          private String keyspace;
  
          @Parameters(index = "1..*", description = "The tables", arity = "0..*")
          private List<String> tables;
  
          protected void execute(NodeProbe probe)
          {
              schemaArgs = concatArgs(keyspace, tables);
              PrintStream out = probe.output().out;
              out.println("Cleaning up data from completed sessions...");
              List<CompositeData> compositeData = probe.getRepairServiceProxy().cleanupPending(schemaArgs, getRangeString(startToken, endToken), force);
  
              List<CleanupSummary> summaries = new ArrayList<>(compositeData.size());
              compositeData.forEach(cd -> summaries.add(CleanupSummary.fromComposite(cd)));
  
              summaries.sort((l, r) -> {
                  int cmp = l.keyspace.compareTo(r.keyspace);
                  if (cmp != 0)
                      return cmp;
  
                  return l.table.compareTo(r.table);
              });
  
              List<String> header = Lists.newArrayList("keyspace", "table", "successful sessions", "unsuccessful sessions");
              List<List<String>> rows = new ArrayList<>(summaries.size() + 1);
              rows.add(header);
  
              boolean hasFailures = false;
              for (CleanupSummary summary : summaries)
              {
                  List<String> row = Lists.newArrayList(summary.keyspace,
                                                        summary.table,
                                                        Integer.toString(summary.successful.size()),
                                                        Integer.toString(summary.unsuccessful.size()));
  
                  hasFailures |= !summary.unsuccessful.isEmpty();
                  rows.add(row);
              }
  
              if (hasFailures)
                  out.println("Some tables couldn't be cleaned up completely");
  
              printTable(rows, out);
          }
      }
  
      @Command(name = "cancel", description = "cancel an incremental repair session." +
                                              " Use --force to cancel from a node other than the repair coordinator" +
                                              " Attempting to cancel FINALIZED or FAILED sessions is an error.")
      public static class CancelCmd extends AbstractCommand
      {
          @Option(paramLabel = "force", names = { "-f", "--force" }, description = "Force a cancellation.")
          private boolean force = false;
  
          @Option(paramLabel = "session", names = { "-s", "--session" }, description = "The session to cancel", required = true)
          private String sessionToCancel;
  
          protected void execute(NodeProbe probe)
          {
              probe.getRepairServiceProxy().failSession(sessionToCancel, force);
          }
      }
  
      private static void printTable(List<List<String>> rows, PrintStream out)
      {
          if (rows.isEmpty())
              return;
  
          // get max col widths
          int[] widths = new int[rows.get(0).size()];
          for (List<String> row : rows)
          {
              assert row.size() == widths.length;
              for (int i = 0; i < widths.length; i++)
              {
                  widths[i] = Math.max(widths[i], row.get(i).length());
              }
          }
  
          List<String> fmts = new ArrayList<>(widths.length);
          for (int i = 0; i < widths.length; i++)
          {
              fmts.add("%-" + widths[i] + "s");
          }
  
          // print
          for (List<String> row : rows)
          {
              List<String> formatted = new ArrayList<>(row.size());
              for (int i = 0; i < widths.length; i++)
              {
                  formatted.add(String.format(fmts.get(i), row.get(i)));
              }
              out.println(Joiner.on(" | ").join(formatted));
          }
      }
  
      static String getRangeString(String startToken, String endToken)
      {
          String rangeStr = null;
          if (!startToken.isEmpty() || !endToken.isEmpty())
              rangeStr = startToken + ':' + endToken;
          return rangeStr;
      }
  }
  