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.service.disk.usage;
  
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.TimeUnit;
  
  import com.google.common.annotations.VisibleForTesting;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.gms.ApplicationState;
  import org.apache.cassandra.gms.EndpointState;
  import org.apache.cassandra.gms.Gossiper;
  import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
  import org.apache.cassandra.gms.VersionedValue;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.service.StorageService;
  import org.apache.cassandra.utils.NoSpamLogger;
  
  /**
   * Starts {@link DiskUsageMonitor} to monitor local disk usage state and broadcast new state via Gossip.
   * At the same time, it caches cluster's disk usage state received via Gossip.
   */
  public class DiskUsageBroadcaster implements IEndpointStateChangeSubscriber
  {
      private static final Logger logger = LoggerFactory.getLogger(DiskUsageBroadcaster.class);
      private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 10, TimeUnit.MINUTES);
  
      public static final DiskUsageBroadcaster instance = new DiskUsageBroadcaster(DiskUsageMonitor.instance);
  
      private final DiskUsageMonitor monitor;
      private final ConcurrentMap<InetAddressAndPort, DiskUsageState> usageInfo = new ConcurrentHashMap<>();
      private volatile boolean hasStuffedOrFullNode = false;
  
      @VisibleForTesting
      public DiskUsageBroadcaster(DiskUsageMonitor monitor)
      {
          this.monitor = monitor;
          // TODO: switch to TCM?
          Gossiper.instance.register(this);
      }
  
      /**
       * @return {@code true} if any node in the cluster is STUFFED OR FULL
       */
      public boolean hasStuffedOrFullNode()
      {
          return hasStuffedOrFullNode;
      }
  
      /**
       * @return {@code true} if given node's disk usage is FULL
       */
      public boolean isFull(InetAddressAndPort endpoint)
      {
          return state(endpoint).isFull();
      }
  
      /**
       * @return {@code true} if given node's disk usage is STUFFED
       */
      public boolean isStuffed(InetAddressAndPort endpoint)
      {
          return state(endpoint).isStuffed();
      }
  
      @VisibleForTesting
      public DiskUsageState state(InetAddressAndPort endpoint)
      {
          return usageInfo.getOrDefault(endpoint, DiskUsageState.NOT_AVAILABLE);
      }
  
      public void startBroadcasting()
      {
          monitor.start(newState -> {
              logger.trace("Disseminating disk usage info: {}", newState);
              Gossiper.instance.addLocalApplicationState(ApplicationState.DISK_USAGE,
                                                         StorageService.instance.valueFactory.diskUsage(newState.name()));
          });
      }
  
      @Override
      public void onChange(InetAddressAndPort endpoint, ApplicationState state, VersionedValue value)
      {
          if (state != ApplicationState.DISK_USAGE)
              return;
  
          DiskUsageState usageState = DiskUsageState.NOT_AVAILABLE;
          try
          {
              usageState = DiskUsageState.valueOf(value.value);
          }
          catch (IllegalArgumentException e)
          {
              noSpamLogger.warn(String.format("Found unknown DiskUsageState: %s. Using default state %s instead.",
                                              value.value, usageState));
          }
          usageInfo.put(endpoint, usageState);
  
          hasStuffedOrFullNode = usageState.isStuffedOrFull() || computeHasStuffedOrFullNode();
      }
  
      private boolean computeHasStuffedOrFullNode()
      {
          for (DiskUsageState replicaState : usageInfo.values())
          {
              if (replicaState.isStuffedOrFull())
              {
                  return true;
              }
          }
          return false;
      }
  
      @Override
      public void onJoin(InetAddressAndPort endpoint, EndpointState epState)
      {
          updateDiskUsage(endpoint, epState);
      }
  
      @Override
      public void beforeChange(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue)
      {
          // nothing to do here
      }
  
      @Override
      public void onAlive(InetAddressAndPort endpoint, EndpointState state)
      {
          updateDiskUsage(endpoint, state);
      }
  
      @Override
      public void onDead(InetAddressAndPort endpoint, EndpointState state)
      {
          // do nothing, as we don't care about dead nodes
      }
  
      @Override
      public void onRestart(InetAddressAndPort endpoint, EndpointState state)
      {
          updateDiskUsage(endpoint, state);
      }
  
      @Override
      public void onRemove(InetAddressAndPort endpoint)
      {
          usageInfo.remove(endpoint);
          hasStuffedOrFullNode = usageInfo.values().stream().anyMatch(DiskUsageState::isStuffedOrFull);
      }
  
      private void updateDiskUsage(InetAddressAndPort endpoint, EndpointState state)
      {
          VersionedValue localValue = state.getApplicationState(ApplicationState.DISK_USAGE);
  
          if (localValue != null)
          {
              onChange(endpoint, ApplicationState.DISK_USAGE, localValue);
          }
      }
  }
  