package com.atlassian.bitbucket.mesh.throttle;

import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.platform.OperatingSystemService;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import java.time.Clock;
import java.util.OptionalDouble;
import java.util.OptionalLong;
import java.util.concurrent.TimeUnit;

import static com.atlassian.bitbucket.mesh.throttle.ThrottledResource.PACK;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.Silent.class)
public class AdaptiveResourceThrottleStrategyTest {

    private static final OptionalDouble CPU_UNAVAILABLE = OptionalDouble.empty();

    private static final int INTERVAL_SEC = 10;
    private static final double TARGET_CPU = 0.75d;
    // assume 8 cores for all test envs to keep consistent
    private static final double CPU_PER_TICKET = 0.25d / 8;
    private static final double PAST_CPU_WEIGHTING = 0.33d;
    private static final int TICKET_MAX = 100;
    private static final int TICKET_MIN = 15;
    private static final int TICKET_SAFE = TICKET_MIN * 2;
    private static final int TIMEOUT_SEC = 25;
    private static final long TIME_START = 1474626308L;
    private static final int FALLBACK_DELAY_SEC = INTERVAL_SEC * 2;

    @Rule
    public final LogListener logListener = new LogListener(AdaptiveResourceThrottleStrategy.class);

    @Mock
    private SemaphoreTicketBucket bucket;
    @Mock
    private Clock clock;
    @Mock
    private OperatingSystemService operatingSystemService;
    @Mock
    private TicketSummary summary;

    @Before
    public void setUp() {
        when(clock.millis()).thenReturn(TIME_START);
        when(bucket.summarize()).thenReturn(summary);
        when(bucket.getTotalTickets()).thenReturn(TICKET_MIN);
        when(operatingSystemService.getTotalPhysicalMemory()).thenReturn(OptionalLong.of(Long.MAX_VALUE));
    }

    @Test
    public void testCreateBucket() {
        SemaphoreTicketBucket bucket = createStrategy().create();

        assertNotNull(bucket);
        assertEquals(TimeUnit.SECONDS.toMillis(TIMEOUT_SEC), bucket.getDefaultAcquireTimeoutMillis());
        assertEquals(TICKET_SAFE, bucket.getTotalTickets());
    }

    @Test
    public void testGetUpdateInterval() {
        assertEquals(OptionalLong.of(INTERVAL_SEC), createStrategy().getUpdateInterval(TimeUnit.SECONDS));
    }

    @Test
    public void testUpdateDecreasesTicketsUnderLoad() {
        AdaptiveResourceThrottleStrategy strategy = createStrategy(TICKET_MAX);

        int curMaxTickets = TICKET_MAX;
        for (int i = 0; i < 1000 && curMaxTickets > TICKET_MIN; ++i) { // to ensure the loop ends
            // all tickets in use
            interval(i + 1, curMaxTickets, curMaxTickets, OptionalDouble.of(TARGET_CPU + 0.05d)); // 5% over target cpu

            strategy.update(bucket);

            ArgumentCaptor<Integer> ticketLimit = ArgumentCaptor.forClass(Integer.class);
            verify(bucket, times(i + 1)).setAcquireTimeoutMillis(strategy.getAcquireTimeoutSeconds() * 1000L);
            verify(bucket, times(i + 1)).setPermits(ticketLimit.capture());
            int prevMaxTickets = curMaxTickets;
            curMaxTickets = ticketLimit.getValue();

            // verify every new ticket max is smaller than the previous (>= to account for rounding)
            assertTrue(prevMaxTickets >= curMaxTickets);
        }
    }

    @Test
    public void testUpdateFallsBackToFixedTicketingAfterNoCpuLoadFromBeginning() {
        AdaptiveResourceThrottleStrategy strategy = createStrategy(TICKET_SAFE);

        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        assertTrue(strategy.isAdapting());

        interval(0, TICKET_SAFE, TICKET_SAFE, OptionalDouble.empty());
        strategy.update(bucket);
        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        assertTrue(strategy.isPaused());

        interval(1, TICKET_SAFE, TICKET_SAFE, OptionalDouble.empty());
        strategy.update(bucket);
        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        assertTrue(strategy.isPaused());

        interval(2, TICKET_SAFE, TICKET_SAFE, OptionalDouble.empty());
        strategy.update(bucket);
        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        assertTrue(strategy.isDisabled());

        interval(3, TICKET_SAFE, TICKET_SAFE, OptionalDouble.empty());
        strategy.update(bucket);
        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        assertTrue(strategy.isDisabled());

        verify(bucket, never()).setPermits(anyInt());
        logListener.assertWarn("A fixed ticket limit of " + TICKET_SAFE +
                " is in place until the reading is next available");
    }

    @Test
    public void testUpdateHandlesCpuLoadNotAvailable() {
        AdaptiveResourceThrottleStrategy strategy = createStrategy();

        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        assertTrue(strategy.isAdapting());

        InOrder inOrder = inOrder(bucket);

        // CPU not available at the start

        interval(0, TICKET_MIN, TICKET_SAFE, CPU_UNAVAILABLE);
        strategy.update(bucket);
        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        inOrder.verify(bucket, never()).setPermits(anyInt());
        assertTrue(strategy.isPaused());

        // CPU suddenly available

        interval(1, TICKET_MIN, TICKET_SAFE, OptionalDouble.of(TARGET_CPU / 2.0));
        strategy.update(bucket);
        assertEquals(27, strategy.getTotalTickets());
        inOrder.verify(bucket).setPermits(27);
        assertTrue(strategy.isAdapting());

        // CPU unavailable for 3 intervals

        interval(2, TICKET_MIN, 27, CPU_UNAVAILABLE);
        strategy.update(bucket);
        assertEquals(27, strategy.getTotalTickets());
        inOrder.verify(bucket, never()).setPermits(anyInt());
        assertTrue(strategy.isPaused());

        interval(3, TICKET_MIN, 27, CPU_UNAVAILABLE);
        strategy.update(bucket);
        assertEquals(27, strategy.getTotalTickets());
        inOrder.verify(bucket, never()).setPermits(anyInt());
        assertTrue(strategy.isPaused());

        interval(4, TICKET_MIN, 27, CPU_UNAVAILABLE);
        strategy.update(bucket);
        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        inOrder.verify(bucket).setPermits(TICKET_SAFE);
        inOrder.verify(bucket, never()).setPermits(anyInt());
        assertTrue(strategy.isDisabled());

        logListener.assertWarn("A fixed ticket limit of " + TICKET_SAFE +
                " is in place until the reading is next available");

        // CPU suddenly available again

        interval(5, TICKET_MIN, TICKET_SAFE, OptionalDouble.of(TARGET_CPU / 2.0));
        strategy.update(bucket);
        assertEquals(27, strategy.getTotalTickets());
        inOrder.verify(bucket).setPermits(27);
        assertTrue(strategy.isAdapting());
    }

    @Test
    public void testUpdateHandlesZeroCpuLoad() {
        AdaptiveResourceThrottleStrategy strategy = mockCpuAndCreate(0.0);

        assertEquals(TICKET_SAFE, strategy.getTotalTickets());
        assertTrue(strategy.isAdapting());

        interval(0, 0, TICKET_SAFE, OptionalDouble.of(0.0));
        strategy.update(bucket);
        verify(bucket).setPermits((int) Math.round(0.75d / CPU_PER_TICKET));
        assertTrue(strategy.isAdapting());

        interval(1, 24, 24, OptionalDouble.of(0.0));
        strategy.update(bucket);
        verify(bucket).setPermits(24 + (int) Math.round(0.75d / CPU_PER_TICKET));
        assertTrue(strategy.isAdapting());
    }

    @Test
    public void testUpdateRespectsLowerLimit() {
        AdaptiveResourceThrottleStrategy strategy = createStrategy();

        // simulate bucket is already at lower ticket limit
        interval(0, TICKET_MIN, TICKET_SAFE, OptionalDouble.of(1.0d));

        strategy.update(bucket);

        // verify that we stay exactly on the limit
        verify(bucket).setPermits(TICKET_MIN);
    }

    @Test
    public void testUpdateSetsSafeTicketCountOnException() {
        when(bucket.summarize())
            .thenReturn(summary)
            .thenThrow(new RuntimeException("Mocking exception from summarize()"));

        AdaptiveResourceThrottleStrategy strategy = createStrategy();

        interval(0, 0, TICKET_SAFE, OptionalDouble.of(0.25d));
        strategy.update(bucket);
        verify(bucket).setPermits(16);

        interval(0, 0, 15, OptionalDouble.of(0.5d));
        strategy.update(bucket);

        // verify the safe/initial ticket level is set
        verify(bucket).setPermits(TICKET_SAFE);

        // although we are in error we'll still try to adapt next round
        assertTrue(strategy.isAdapting());

        logListener.assertError("Error encountered during adaptive throttling update. Setting tickets to " +
                TICKET_SAFE + " as a fallback measure");
    }

    @Test
    public void testPastCPUUsageCorrectlyWeighted() {
        // target 60% cpu usage, past CPU usage given 50% weighting
        AdaptiveResourceThrottleStrategy
                strategy = new AdaptiveResourceThrottleStrategy(PACK, clock,
                operatingSystemService, INTERVAL_SEC, TICKET_MIN, TICKET_MAX, TICKET_SAFE, 0.6d, CPU_PER_TICKET, 0.5d,
                TIMEOUT_SEC, FALLBACK_DELAY_SEC);

        InOrder inOrder = inOrder(bucket);

        interval(0, TICKET_SAFE, TICKET_SAFE, OptionalDouble.of(0d));
        strategy.update(bucket);
        inOrder.verify(bucket).setPermits(49);

        interval(1, 19, 49, OptionalDouble.of(0.4d));
        strategy.update(bucket);
        inOrder.verify(bucket).setPermits(32);

        interval(2, 25, 32, OptionalDouble.of(0.2d));
        strategy.update(bucket);
        inOrder.verify(bucket).setPermits(38);

        interval(3, 30, 38, OptionalDouble.of(0.9d));
        strategy.update(bucket);
        inOrder.verify(bucket).setPermits(32);

        interval(4, 0, 32, OptionalDouble.of(0.5d));
        strategy.update(bucket);
        inOrder.verify(bucket).setPermits(15);
    }

    @Test
    public void testUpdateTicketIncreaseIsCappedByTicketMax() {
        // initialize the strategy just under MAX tickets
        AdaptiveResourceThrottleStrategy strategy = createStrategy(TICKET_MAX - 1);

        // mock that all tickets are in use and CPU is still only at 25%
        interval(1, TICKET_MAX - 1, TICKET_MAX - 1, OptionalDouble.of(0.25d));

        strategy.update(bucket);

        // validate that the ticket growth is capped by TICKET_MAX
        verify(bucket).setPermits(TICKET_MAX);
        verify(bucket).setAcquireTimeoutMillis(anyLong()); // acquire timeout is always updated when update is called

        // mock another interval where all tickets are in use and CPU is still only at 25%
        interval(2, TICKET_MAX, TICKET_MAX, OptionalDouble.of(0.25d));

        strategy.update(bucket);

        // validate that the limit is not increased beyond TICKET_MAX
        verify(bucket, times(2)).summarize();
        verify(bucket, times(2)).setAcquireTimeoutMillis(anyLong()); // acquire timeout should be update again
        verifyNoMoreInteractions(bucket);
    }

    private AdaptiveResourceThrottleStrategy mockCpuAndCreate(double cpuLoad) {
        // recreate the strategy so it gets
        when(operatingSystemService.getCpuLoad()).thenReturn(OptionalDouble.of(cpuLoad));

        return createStrategy();
    }

    private AdaptiveResourceThrottleStrategy createStrategy() {
        return createStrategy(TICKET_SAFE);
    }

    private AdaptiveResourceThrottleStrategy createStrategy(int initialTickets) {
        return new AdaptiveResourceThrottleStrategy(PACK, clock, operatingSystemService,
                INTERVAL_SEC, TICKET_MIN, TICKET_MAX, initialTickets, TARGET_CPU, CPU_PER_TICKET, PAST_CPU_WEIGHTING,
                TIMEOUT_SEC, FALLBACK_DELAY_SEC);
    }

    private void interval(int index, int ticketsUsed, int ticketsTotal, OptionalDouble cpu) {
        when(summary.getUsed()).thenReturn(ticketsUsed);
        when(summary.getTotal()).thenReturn(ticketsTotal);
        when(operatingSystemService.getCpuLoad()).thenReturn(cpu);

        // tick + 1 because millis() is called from the ctor of the strategy and will return TIME_START
        when(clock.millis()).thenReturn(TIME_START + (index + 1) * TimeUnit.SECONDS.toMillis(INTERVAL_SEC));
    }
}
