providers/openflow/flow/src/main/java/org/onosproject/provider/of/flow/impl/FlowStatsCollector.java

/*
 * Copyright 2014-present Open Networking Foundation
 *
 * Licensed 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.onosproject.provider.of.flow.impl;

import static com.google.common.base.Preconditions.checkNotNull;

import org.onlab.util.SlidingWindowCounter;
import org.onosproject.openflow.controller.OpenFlowSwitch;
import org.onosproject.openflow.controller.RoleState;
import org.projectfloodlight.openflow.protocol.OFFlowStatsRequest;
import org.projectfloodlight.openflow.types.OFPort;
import org.projectfloodlight.openflow.types.TableId;
import org.slf4j.Logger;

import java.util.TimerTask;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static java.lang.Math.max;
import static java.lang.Math.min;
import static org.slf4j.LoggerFactory.getLogger;

/**
 * Collects flow statistics for the specified switch.
 */
class FlowStatsCollector implements SwitchDataCollector {

    private final Logger log = getLogger(getClass());

    private static final int MS = 1000;

    // Number of ticks which defines the pause window.
    private static final int PAUSE_WINDOW = 2;
    // Number of ticks which defines the high load window
    private static final int HIGH_WINDOW = 60;
    // Number of ticks which defines the low load window
    private static final int LOW_WINDOW = 15;
    // Multiplier factor of the load
    private static final int LOAD_FACTOR = 2;
    // Event/s defining the min load rate
    private static final int MIN_LOAD_RATE = 50;
    // Event/s defining the max load rate
    private static final int MAX_LOAD_RATE = 500;

    private final OpenFlowSwitch sw;
    private ScheduledExecutorService executorService;
    private TimerTask pauseTask;
    private ScheduledFuture<?> scheduledPauseTask;
    private TimerTask pollTask;
    private ScheduledFuture<?> scheduledPollTask;

    private SlidingWindowCounter loadCounter;
    // Defines whether the collector is in pause or not for high load
    private final AtomicBoolean paused = new AtomicBoolean();
    // Defines whether the collector is in waiting or not for a previous stats reply
    private static final int WAITING_ATTEMPTS = 5;
    private final AtomicInteger waiting = new AtomicInteger(0);

    private int pollInterval;

    /**
     * Creates a new collector for the given switch and poll frequency.
     *
     * @param executorService executor used for scheduling
     * @param sw switch to pull
     * @param pollInterval poll frequency in seconds
     */
    FlowStatsCollector(ScheduledExecutorService executorService, OpenFlowSwitch sw, int pollInterval) {
        this.executorService = executorService;
        this.sw = checkNotNull(sw, "Null switch");
        this.pollInterval = pollInterval;
    }

    /**
     * Adjusts poll frequency.
     *
     * @param pollInterval poll frequency in seconds
     */
    synchronized void adjustPollInterval(int pollInterval) {
        this.pollInterval = pollInterval;
        if (pollTask != null) {
            pollTask.cancel();
        }
        if (scheduledPollTask != null) {
            scheduledPollTask.cancel(false);
        }
        // If we went through start - let's schedule it
        if (loadCounter != null) {
            pollTask = new PollTimerTask();
            scheduledPollTask = executorService.scheduleAtFixedRate(pollTask, pollInterval * MS,
                                                pollInterval * MS, TimeUnit.MILLISECONDS);
        }
        waiting.set(0);
    }

    /**
     * Resets the collector's event count.
     */
    @Override
    public synchronized void resetEvents() {
        loadCounter.clear();
        if (paused.compareAndSet(true, false)) {
            resume();
        }
        // Let's reset also waiting, the reply can be discarded/lost
        // during a change of mastership
        waiting.set(0);
    }

    /**
     * Records a number of flow events that have occurred.
     *
     * @param events the number of events that occurred
     */
    @Override
    public void recordEvents(int events) {
        SlidingWindowCounter loadCounter = this.loadCounter;
        if (loadCounter != null) {
            loadCounter.incrementCount(events);
        }
    }

    /**
     * Returns a boolean indicating whether the switch is under high load.
     * <p>
     * The switch is considered under high load if the average rate over the last two seconds is
     * greater than twice the overall rate or 50 flows/sec.
     *
     * @return indicates whether the switch is under high load
     */
    private boolean isHighLoad() {
        return loadCounter.getWindowRate(PAUSE_WINDOW)
            > max(min(loadCounter.getWindowRate(HIGH_WINDOW) * LOAD_FACTOR, MAX_LOAD_RATE), MIN_LOAD_RATE);
    }

    /**
     * Returns a boolean indicating whether the switch is under low load.
     * <p>
     * The switch is considered under low load if the average rate over the last 15 seconds is
     * less than the overall rate.
     *
     * @return indicates whether the switch is under low load
     */
    private boolean isLowLoad() {
        return loadCounter.getWindowRate(LOW_WINDOW) < loadCounter.getWindowRate(HIGH_WINDOW);
    }

    private class PauseTimerTask extends TimerTask {
        @Override
        public void run() {
            if (isHighLoad()) {
                if (paused.compareAndSet(false, true)) {
                    pause();
                }
            } else if (isLowLoad()) {
                if (paused.compareAndSet(true, false)) {
                    resume();
                }
            }
        }
    }

    private class PollTimerTask extends TimerTask {
        @Override
        public void run() {
            // Check whether we are still waiting a previous reply
            if (waiting.getAndDecrement() > 0) {
                log.debug("Skipping stats collection for {} waiting for previous reply", sw.getStringId());
                return;
            }
            // Check whether we are the master of the switch
            if (sw.getRole() == RoleState.MASTER) {
                // Check whether the switch is under high load from this master. This is done here in case a large
                // batch was pushed immediately prior to this task running.
                if (isHighLoad()) {
                    log.debug("Skipping stats collection for {} due to high load; rate: {}; overall: {}",
                              sw.getStringId(),
                              loadCounter.getWindowRate(PAUSE_WINDOW),
                              loadCounter.getWindowRate(HIGH_WINDOW));
                    return;
                } else {
                    log.debug(
                        "Permitting stats collection for {}; rate: {}; overall: {}",
                        sw.getStringId(),
                        loadCounter.getWindowRate(PAUSE_WINDOW),
                        loadCounter.getWindowRate(HIGH_WINDOW));
                }

                log.trace("Collecting stats for {}", sw.getStringId());
                OFFlowStatsRequest request = sw.factory().buildFlowStatsRequest()
                        .setMatch(sw.factory().matchWildcardAll())
                        .setTableId(TableId.ALL)
                        .setOutPort(OFPort.NO_MASK)
                        .build();
                sw.sendMsg(request);
                // Other flow stats will not be asked
                // if we don't see first the reply of this request
                waiting.set(WAITING_ATTEMPTS);
            }
        }
    }

    public synchronized void start() {
        log.debug("Starting Stats collection thread for {}", sw.getStringId());
        loadCounter = new SlidingWindowCounter(HIGH_WINDOW);
        if (pollInterval > 0) {
            pauseTask = new PauseTimerTask();
            scheduledPauseTask = executorService.scheduleAtFixedRate(pauseTask, 1 * MS,
                    1 * MS, TimeUnit.MILLISECONDS);
            pollTask = new PollTimerTask();
            // Initially start polling quickly. Then drop down to configured value
            scheduledPollTask = executorService.scheduleAtFixedRate(pollTask, 1 * MS,
                    pollInterval * MS, TimeUnit.MILLISECONDS);
        } else {
            // Trigger the poll only once
            pollTask = new PollTimerTask();
            executorService.schedule(pollTask, 0, TimeUnit.MILLISECONDS);
        }
    }

    private synchronized void pause() {
        if (pollTask != null) {
            log.debug("Pausing stats collection for {}; rate: {}; overall: {}",
                      sw.getStringId(),
                      loadCounter.getWindowRate(PAUSE_WINDOW),
                      loadCounter.getWindowRate(HIGH_WINDOW));
            pollTask.cancel();
            pollTask = null;
        }
        if (scheduledPollTask != null) {
            scheduledPollTask.cancel(false);
            scheduledPollTask = null;
        }
    }

    private synchronized void resume() {
        log.debug("Resuming stats collection for {}; rate: {}; overall: {}",
                  sw.getStringId(),
                  loadCounter.getWindowRate(PAUSE_WINDOW),
                  loadCounter.getWindowRate(HIGH_WINDOW));
        pollTask = new PollTimerTask();
        scheduledPollTask = executorService.scheduleAtFixedRate(pollTask, pollInterval * MS,
                                            pollInterval * MS, TimeUnit.MILLISECONDS);
    }

    public synchronized void stop() {
        if (pauseTask != null) {
            pauseTask.cancel();
            pauseTask = null;
        }
        if (scheduledPauseTask != null) {
            scheduledPauseTask.cancel(false);
            scheduledPauseTask = null;
        }
        if (pollTask != null) {
            log.debug("Stopping Stats collection thread for {}", sw.getStringId());
            pollTask.cancel();
            pollTask = null;
        }
        if (scheduledPollTask != null) {
            scheduledPollTask.cancel(false);
            scheduledPollTask = null;
        }
        if (loadCounter != null) {
            loadCounter.destroy();
            loadCounter = null;
        }
    }

    public void received() {
        waiting.set(0);
    }

}