<commit_before>/* Copyright 2018 Telstra Open Source
 *
 *   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.openkilda.wfm.topology.ping.bolt;

import org.openkilda.wfm.error.AbstractException;
import org.openkilda.wfm.error.PipelineException;
import org.openkilda.wfm.topology.ping.model.PingContext;
import org.openkilda.wfm.topology.ping.model.PingContext.Kinds;
import org.openkilda.wfm.topology.ping.model.ShapingBurstCalculator;

import lombok.extern.slf4j.Slf4j;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.LinkedList;

@Slf4j
public class PeriodicPingShaping extends Abstract {
    public static final String BOLT_ID = ComponentId.PERIODIC_PING_SHAPING.toString();

    public static final String FIELD_ID_FLOW_ID = FlowFetcher.FIELD_ID_FLOW_ID;
    public static final Fields STREAM_FIELDS = new Fields(FIELD_ID_FLOW_ID, FIELD_ID_PING, FIELD_ID_CONTEXT);

    private final int windowSize;
    private ShapingBurstCalculator burstCalculator = null;

    private long proxyCounter = 0;
    private long proxyAllowance = 0;
    private LinkedList<Values> backlog = null;

    public PeriodicPingShaping(int pingInterval) {
        int window = pingInterval * 5;
        window += pingInterval / 2;
        window += pingInterval % 2 > 0 ? 1 : 0;
        this.windowSize =  window;
    }

    @Override
    protected void handleInput(Tuple input) throws AbstractException {
        String source = input.getSourceComponent();

        if (FlowFetcher.BOLT_ID.equals(source)) {
            handlePing(input);
        } else if (TickDeduplicator.BOLT_ID.equals(source)) {
            handleMonotonicTick(input);
        } else {
            unhandledInput(input);
        }
    }

    private void handlePing(Tuple input) throws PipelineException {
        Values output = extract(input);

        if (burstCalculator == null) {
            // silent proxy record, due to disabled shaping
            emit(input, output);
            return;
        }

        PingContext ping = pullPingContext(input);
        if (ping.getKind() != Kinds.PERIODIC) {
            emit(input, output);
        }

        burstCalculator.increment();
        proxy(input, output, ping);
        logStats();
    }

    private void handleMonotonicTick(Tuple input) {
        burstCalculator.slide();
        proxyAllowance = calcProxyAllowance();
        proxyCounter = 0;

        log.debug("Emit ping records from backlog");
        releaseBurst(input);<commit_msg>Avoid extra on-demand ping requests<commit_after>/* Copyright 2018 Telstra Open Source
 *
 *   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.openkilda.wfm.topology.ping.bolt;

import org.openkilda.wfm.error.AbstractException;
import org.openkilda.wfm.error.PipelineException;
import org.openkilda.wfm.topology.ping.model.PingContext;
import org.openkilda.wfm.topology.ping.model.PingContext.Kinds;
import org.openkilda.wfm.topology.ping.model.ShapingBurstCalculator;

import lombok.extern.slf4j.Slf4j;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.LinkedList;

@Slf4j
public class PeriodicPingShaping extends Abstract {
    public static final String BOLT_ID = ComponentId.PERIODIC_PING_SHAPING.toString();

    public static final String FIELD_ID_FLOW_ID = FlowFetcher.FIELD_ID_FLOW_ID;
    public static final Fields STREAM_FIELDS = new Fields(FIELD_ID_FLOW_ID, FIELD_ID_PING, FIELD_ID_CONTEXT);

    private final int windowSize;
    private ShapingBurstCalculator burstCalculator = null;

    private long proxyCounter = 0;
    private long proxyAllowance = 0;
    private LinkedList<Values> backlog = null;

    public PeriodicPingShaping(int pingInterval) {
        int window = pingInterval * 5;
        window += pingInterval / 2;
        window += pingInterval % 2 > 0 ? 1 : 0;
        this.windowSize =  window;
    }

    @Override
    protected void handleInput(Tuple input) throws AbstractException {
        String source = input.getSourceComponent();

        if (FlowFetcher.BOLT_ID.equals(source)) {
            handlePing(input);
        } else if (TickDeduplicator.BOLT_ID.equals(source)) {
            handleMonotonicTick(input);
        } else {
            unhandledInput(input);
        }
    }

    private void handlePing(Tuple input) throws PipelineException {
        Values output = extract(input);

        if (burstCalculator == null) {
            // ping-shaping disabled, silently proxying
            emit(input, output);
            return;
        }

        PingContext ping = pullPingContext(input);
        if (ping.getKind() != Kinds.PERIODIC) {
            emit(input, output);
            return;
        }

        burstCalculator.increment();
        proxy(input, output, ping);
        logStats();
    }

    private void handleMonotonicTick(Tuple input) {
        if (burstCalculator == null) {
            return;
        }

        burstCalculator.slide();
        proxyAllowance = calcProxyAllowance();
        proxyCounter = 0;

        log.debug("Emit ping records from backlog");
        releaseBurst(input);
{'fraction_non_alphanumeric': 0.08571428571428572, 'fraction_numerical': 0.006279434850863423, 'mean_word_length': 3.0195583596214512, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4828151', 'n_tokens_mistral': 1927, 'n_tokens_neox': 1836, 'n_words': 489}