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.simulator;
  
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.EnumMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  import java.util.function.Consumer;
  import java.util.function.Function;
  import java.util.function.Predicate;
  import java.util.stream.Collectors;
  import javax.annotation.Nullable;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.BufferDecoratedKey;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.Keyspace;
  import org.apache.cassandra.db.marshal.Int32Type;
  import org.apache.cassandra.dht.Murmur3Partitioner;
  import org.apache.cassandra.distributed.Cluster;
  import org.apache.cassandra.distributed.api.IIsolatedExecutor.TriFunction;
  import org.apache.cassandra.gms.EndpointState;
  import org.apache.cassandra.gms.Gossiper;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.locator.ReplicaLayout;
  import org.apache.cassandra.schema.Schema;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.simulator.systems.SimulatedTime;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.utils.FBUtilities;
  
  import static java.util.function.Function.identity;
  import static org.apache.cassandra.simulator.Action.Modifier.INFO;
  import static org.apache.cassandra.simulator.Action.Modifier.WAKEUP;
  import static org.apache.cassandra.simulator.ActionListener.recursive;
  import static org.apache.cassandra.simulator.ActionListener.runAfter;
  import static org.apache.cassandra.simulator.ActionListener.runAfterAndTransitivelyAfter;
  import static org.apache.cassandra.simulator.Debug.EventType.CLUSTER;
  import static org.apache.cassandra.simulator.Debug.EventType.PARTITION;
  import static org.apache.cassandra.simulator.Debug.Info.LOG;
  import static org.apache.cassandra.simulator.Debug.Level.PLANNED;
  import static org.apache.cassandra.simulator.paxos.Ballots.paxosDebugInfo;
  
  // TODO (feature): move logging to a depth parameter
  // TODO (feature): log only deltas for schema/cluster data
  public class Debug
  {
      private static class LoggerHandle
      {
          private static final Logger logger = LoggerFactory.getLogger(Debug.class);
      }
  
      private static Logger logger()
      {
          return LoggerHandle.logger;
      }
  
      public enum EventType { PARTITION, CLUSTER }
      public enum Level
      {
          PLANNED,
          CONSEQUENCES,
          ALL;
  
          private static final Level[] LEVELS = values();
      }
      public enum Info
      {
          LOG(EventType.values()),
          PAXOS(PARTITION),
          OWNERSHIP(CLUSTER),
          GOSSIP(CLUSTER),
          RF(CLUSTER),
          RING(CLUSTER);
  
          public final EventType[] defaultEventTypes;
  
          Info(EventType ... defaultEventTypes)
          {
              this.defaultEventTypes = defaultEventTypes;
          }
      }
  
      public static class Levels
      {
          private final EnumMap<EventType, Level> levels;
  
          public Levels(EnumMap<EventType, Level> levels)
          {
              this.levels = levels;
          }
  
          public Levels(Level level, EventType ... types)
          {
              this.levels = new EnumMap<>(EventType.class);
              for (EventType type : types)
                  this.levels.put(type, level);
          }
  
          public Levels(int partition, int cluster)
          {
              this.levels = new EnumMap<>(EventType.class);
              if (partition > 0) this.levels.put(PARTITION, Level.LEVELS[partition - 1]);
              if (cluster > 0) this.levels.put(CLUSTER, Level.LEVELS[cluster - 1]);
          }
  
          Level get(EventType type)
          {
              return levels.get(type);
          }
  
          boolean anyMatch(Predicate<Level> test)
          {
              return levels.values().stream().anyMatch(test);
          }
      }
  
      private final EnumMap<Info, Levels> levels;
      public final int[] primaryKeys;
  
      public Debug()
      {
          this(new EnumMap<>(Info.class), null);
      }
  
      public Debug(Map<Info, Levels> levels, int[] primaryKeys)
      {
          this.levels = new EnumMap<>(levels);
          this.primaryKeys = primaryKeys;
      }
  
      public ActionListener debug(EventType type, SimulatedTime time, Cluster cluster, String keyspace, Integer primaryKey)
      {
          List<ActionListener> listeners = new ArrayList<>();
          for (Map.Entry<Info, Levels> e : levels.entrySet())
          {
              Info info = e.getKey();
              Level level = e.getValue().get(type);
              if (level == null) continue;
  
              ActionListener listener;
              if (info == LOG)
              {
                  Function<ActionListener, ActionListener> adapt = type == CLUSTER ? LogTermination::new : identity();
                  switch (level)
                  {
                      default: throw new AssertionError();
                      case PLANNED: listener = adapt.apply(new LogOne(time, false)); break;
                      case CONSEQUENCES: case ALL: listener = adapt.apply(recursive(new LogOne(time, true))); break;
                  }
              }
              else if (keyspace != null)
              {
                  Consumer<Action> debug;
                  switch (info)
                  {
                      default: throw new AssertionError();
                      case GOSSIP: debug = debugGossip(cluster); break;
                      case RF: debug = debugRf(cluster, keyspace); break;
                      case RING: debug = debugRing(cluster, keyspace); break;
                      case PAXOS: debug = forKeys(cluster, keyspace, primaryKey, Debug::debugPaxos); break;
                      case OWNERSHIP: debug = forKeys(cluster, keyspace, primaryKey, Debug::debugOwnership); break;
                  }
                  switch (level)
                  {
                      default: throw new AssertionError();
                      case PLANNED: listener = type == CLUSTER ? runAfterAndTransitivelyAfter(debug) : runAfter(debug); break;
                      case CONSEQUENCES: listener = recursive(runAfter(ignoreWakeupAndLogEvents(debug))); break;
                      case ALL: listener = recursive(runAfter(ignoreLogEvents(debug))); break;
                  }
              }
              else continue;
  
              listeners.add(listener);
          }
  
          if (listeners.isEmpty())
              return null;
          return new ActionListener.Combined(listeners);
      }
  
      public boolean isOn(Info info)
      {
          return isOn(info, PLANNED);
      }
  
      public boolean isOn(Info info, Level level)
      {
          Levels levels = this.levels.get(info);
          if (levels == null) return false;
          return levels.anyMatch(test -> level.compareTo(test) >= 0);
      }
  
      @SuppressWarnings("UnnecessaryToStringCall")
      private static class LogOne implements ActionListener
      {
          final SimulatedTime time;
          final boolean logConsequences;
          private LogOne(SimulatedTime time, boolean logConsequences)
          {
              this.time = time;
              this.logConsequences = logConsequences;
          }
  
          @Override
          public void before(Action action, Before before)
          {
              if (logger().isWarnEnabled()) // invoke toString() eagerly to ensure we have the task's descriptin
                  logger().warn(String.format("%6ds %s %s", TimeUnit.NANOSECONDS.toSeconds(time.nanoTime()), before, action));
          }
  
          @Override
          public void consequences(ActionList consequences)
          {
              if (logConsequences && !consequences.isEmpty() && logger().isWarnEnabled())
                  logger().warn(String.format("%6ds Next: %s", TimeUnit.NANOSECONDS.toSeconds(time.nanoTime()), consequences));
          }
      }
  
      private static class LogTermination extends ActionListener.Wrapped
      {
          public LogTermination(ActionListener wrap)
          {
              super(wrap);
          }
  
          @Override
          public void transitivelyAfter(Action finished)
          {
              logger().warn("Terminated {}", finished);
          }
      }
  
      private static Consumer<Action> ignoreWakeupAndLogEvents(Consumer<Action> consumer)
      {
          return action -> {
              if (!action.is(WAKEUP) && !action.is(INFO))
                  consumer.accept(action);
          };
      }
  
      private static Consumer<Action> ignoreLogEvents(Consumer<Action> consumer)
      {
          return action -> {
              if (!action.is(INFO))
                  consumer.accept(action);
          };
      }
  
      private Consumer<Action> debugGossip(Cluster cluster)
      {
          return ignore -> {
              cluster.forEach(i -> i.unsafeRunOnThisThread(() -> {
                  for (InetAddressAndPort ep : Gossiper.instance.getLiveMembers())
                  {
                      EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(ep);
                      logger().warn("Gossip {}: {} {}", ep, epState.isAlive(), epState.states().stream()
                                                                                     .map(e -> e.getKey().toString() + "=(" + e.getValue().value + ',' + e.getValue().version + ')')
                                                                                     .collect(Collectors.joining(", ", "[", "]")));
                  }
              }));
          };
      }
  
      private Consumer<Action> forKeys(Cluster cluster, String keyspace, @Nullable Integer specificPrimaryKey, TriFunction<Cluster, String, Integer, Consumer<Action>> factory)
      {
          if (specificPrimaryKey != null) return factory.apply(cluster, keyspace, specificPrimaryKey);
          else return forEachKey(cluster, keyspace, primaryKeys, Debug::debugPaxos);
      }
  
      public static Consumer<Action> forEachKey(Cluster cluster, String keyspace, int[] primaryKeys, TriFunction<Cluster, String, Integer, Consumer<Action>> factory)
      {
          Consumer<Action>[] eachKey = new Consumer[primaryKeys.length];
          for (int i = 0 ; i < primaryKeys.length ; ++i)
              eachKey[i] = factory.apply(cluster, keyspace, primaryKeys[i]);
  
          return action -> {
              for (Consumer<Action> run : eachKey)
                  run.accept(action);
          };
      }
  
      public static Consumer<Action> debugPaxos(Cluster cluster, String keyspace, int primaryKey)
      {
          return ignore -> {
              for (int node = 1 ; node <= cluster.size() ; ++node)
              {
                  cluster.get(node).unsafeAcceptOnThisThread((num, pkint) -> {
                      try
                      {
                          TableMetadata metadata = Keyspace.open(keyspace).getColumnFamilyStore("tbl").metadata.get();
                          ByteBuffer pkbb = Int32Type.instance.decompose(pkint);
                          DecoratedKey key = new BufferDecoratedKey(DatabaseDescriptor.getPartitioner().getToken(pkbb), pkbb);
                          logger().warn("node{}({}): {}", num, primaryKey, paxosDebugInfo(key, metadata, FBUtilities.nowInSeconds()));
                      }
                      catch (Throwable t)
                      {
                          logger().warn("node{}({})", num, primaryKey, t);
                      }
                  }, node, primaryKey);
              }
          };
      }
  
      public static Consumer<Action> debugRf(Cluster cluster, String keyspace)
      {
          return ignore -> {
              cluster.forEach(i -> i.unsafeRunOnThisThread(() -> {
                  logger().warn("{} {}",
                          Schema.instance.getKeyspaceMetadata(keyspace) == null ? "" : Schema.instance.getKeyspaceMetadata(keyspace).params.replication.toString(),
                          Schema.instance.getKeyspaceMetadata(keyspace) == null ? "" : Keyspace.open(keyspace).getReplicationStrategy().configOptions.toString());
              }));
          };
      }
  
      public static Consumer<Action> debugOwnership(Cluster cluster, String keyspace, int primaryKey)
      {
          return ignore -> {
              for (int node = 1 ; node <= cluster.size() ; ++node)
              {
                  logger().warn("node{}({}): {}", node, primaryKey, cluster.get(node).unsafeApplyOnThisThread(v -> {
                      try
                      {
                          return ReplicaLayout.forTokenWriteLiveAndDown(Keyspace.open(keyspace), Murmur3Partitioner.instance.getToken(Int32Type.instance.decompose(v))).all().endpointList().toString();
                      }
                      catch (Throwable t)
                      {
                          return "Error";
                      }
                  }, primaryKey));
              }
          };
      }
  
      public static Consumer<Action> debugRing(Cluster cluster, String keyspace)
      {
          return ignore -> cluster.forEach(i -> i.unsafeRunOnThisThread(() -> {
              if (Schema.instance.getKeyspaceMetadata(keyspace) != null)
                  logger().warn("{}", ClusterMetadata.current());
          }));
      }
  
  }
  