SamplingAspect.java

package br.ufrgs.inf.prosoft.tigris.sampling;

import br.ufrgs.inf.prosoft.tigris.configuration.annotation.TigrisConfiguration;
import br.ufrgs.inf.prosoft.tigris.exceptions.ConfigurationException;
import br.ufrgs.inf.prosoft.tigris.monitoring.util.threads.NamedThreads;
import br.ufrgs.inf.prosoft.tigris.utils.ConfigurationUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static java.lang.System.nanoTime;

@Aspect
public class SamplingAspect implements Runnable {

    @Pointcut(
            //any execution except the own framework
            "(" +
                    "(execution(* org.dacapo.h2.TPCCSubmitter.runTransaction(..)) || " + //h2
                    "execution(* org.dacapo.lusearch.QueryProcessor.doPagingSearch(..)) || " + //lusearch
                    "execution(* org.dacapo.xalan.XalanWorker.transform(..))) " + //xalan
//            "(execution(* *(..)) && !within(br.ufrgs.inf.prosoft..*) " +
                    //avoid calls from repository while serializing objects, it is necessary if a hash could not be used
                    //"&& !cflow(call(* br.ufrgs.inf.prosoft.tigris.monitoring.storage..*(..))) " +
                    //conditional to enable and disable at runtime
                    "&& if())"
    )
    public static boolean anyCall() {
        return enabled;
    }

    //tigris config
    private final TigrisConfiguration tigrisConfiguration;
    private static Sampling sampling;

    public SamplingAspect() {
        Class<?> configClass = ConfigurationUtils.getAvailableConfigurationClass(TigrisConfiguration.class);
        if (configClass == null) {
            logger.info("Tigris tracing disabled, there is no annotations.");
            throw new ConfigurationException("Tigris tracing disabled, there is no annotations.");
        }

        tigrisConfiguration = configClass.getAnnotation(TigrisConfiguration.class);

        logger.info("@TigrisConfiguration found.");

        sampling = new Sampling(tigrisConfiguration.samplingPercentage(), tigrisConfiguration.cycleTimeInMilliseconds(), tigrisConfiguration.adaptiveSamplingRate());

        adaptiveSamplingExecutor.scheduleWithFixedDelay(
                this::run,
                5, 1, TimeUnit.SECONDS);
    }

    static Logger logger = LoggerFactory.getLogger(SamplingAspect.class);

    public static boolean enabled = false;

    public static boolean samplingEnabled = true;

    private final ScheduledExecutorService adaptiveSamplingExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreads(
            "adaptive-sampling",
            "readiness evaluation, pba trigger and sampling adaptation"
    ));

    @Around("anyCall()")
    public Object aroundMethods(ProceedingJoinPoint joinPoint) throws Throwable {
        long startTime = nanoTime();
        Object result = joinPoint.proceed();
        long endTime = nanoTime();

        if (samplingEnabled) {
            String signature = joinPoint.getSignature().toString() +
                    joinPoint.getArgs()[0].toString(); //TODO this is to distinguish traces, need to change for other than h2 or lusearch
            Granularity granularity = new Granularity(GranularityType.METHOD, signature);
            if (sampling.isPerformanceBaselineEnabled()) {
                sampling.addPerformanceBaselineItem(granularity, endTime - startTime);
                return result;
            }

            boolean decision = sampling.samplingDecision(granularity, endTime - startTime);

            if (decision)
                sampling.addSampledItem(granularity, nanoTime() - startTime);
        }

        return result;
    }

    static double currentSamplingRate;
    static MonitoringCycle cycle = new MonitoringCycle(0,0,0,0);

    @Override
    public void run() {
        if (SamplingAspect.enabled && sampling.isAdaptiveSamplingRate()) {
            sampling.managePerformanceBaseline();
            currentSamplingRate = sampling.getSamplingRate();

            if (!sampling.isPerformanceBaselineEnabled() && sampling.isReady()) {
                logger.info("Sample is ready, releasing for analysis and resetting...");
                cycle = sampling.endMonitoringCycle();
            }
        }
    }

    public static double getCurrentSamplingRate() {
        return currentSamplingRate;
    }

    public static MonitoringCycle getCycle() {
        return cycle;
    }

    public static MonitoringCycle endMonitoringCycle() {
        return sampling.endMonitoringCycle();
    }
}