TigrisCoordinator.java

package br.ufrgs.inf.prosoft.tigris.monitoring.aspects;

import br.ufrgs.inf.prosoft.tigris.configuration.annotation.ComponentScan;
import br.ufrgs.inf.prosoft.tigris.configuration.annotation.TigrisConfiguration;
import br.ufrgs.inf.prosoft.tigris.exceptions.ConfigurationException;
import br.ufrgs.inf.prosoft.tigris.metrics.DataFiltering;
import br.ufrgs.inf.prosoft.tigris.metrics.LightweightMetrics;
import br.ufrgs.inf.prosoft.tigris.metrics.StaticMetrics;
import br.ufrgs.inf.prosoft.tigris.monitoring.metadata.Key;
import br.ufrgs.inf.prosoft.tigris.monitoring.metadata.LogTrace;
import br.ufrgs.inf.prosoft.tigris.monitoring.metadata.MethodInfo;
import br.ufrgs.inf.prosoft.tigris.monitoring.storage.Repository;
import br.ufrgs.inf.prosoft.tigris.monitoring.storage.RepositoryFactory;
import br.ufrgs.inf.prosoft.tigris.monitoring.util.threads.NamedThreads;
import br.ufrgs.inf.prosoft.tigris.sampling.Granularity;
import br.ufrgs.inf.prosoft.tigris.sampling.Sampling;
import br.ufrgs.inf.prosoft.tigris.sampling.SamplingConfiguration;
import br.ufrgs.inf.prosoft.utils.ConfigurationUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

import static java.lang.System.currentTimeMillis;

public class TigrisCoordinator implements Runnable {
    private static boolean enabled = true;
    private boolean coarseMonitoringEnabled = true;

    Logger logger = LoggerFactory.getLogger(TigrisCoordinator.class);

    //traceable configuration
    private Pattern allowedPattern;
    private Pattern deniedPattern;

    //tigris config
    private final TigrisConfiguration tigrisConfiguration;
    private final SamplingConfiguration samplingConfiguration;
    private final Sampling sampling;
    private final Repository repository;
    private Class<?> customMonitoringClass;
    private CustomMonitoring customMonitoring;

    private final String staticFile;
    private Map<String, LightweightMetrics> metrics;

    private final ScheduledExecutorService samplingAdaptationExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreads(
            "sampling-adaptation",
            "adapting sampling rate and sample readiness"
    ));

    private final ScheduledExecutorService lightweightAnalyzerExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreads(
            "lightweight-analyzer",
            "computing lightweight metrics and setting allowed methods"
    ));

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

        tigrisConfiguration = configClass.getAnnotation(TigrisConfiguration.class);
        staticFile = tigrisConfiguration.staticMetricFile();

        //getting allowed packages from @ComponentScan
        ComponentScan componentScanConfig = configClass.getAnnotation(ComponentScan.class);
        if (componentScanConfig == null) {
            //if not specified, it assumes the same package where @AdaptiveCaching were declared
            allowedPattern = Pattern.compile(configClass.getPackage().getName() + ".*");
            logger.info("ComponentScan for TigrisConfiguration not found, assuming the same package where @TigrisConfiguration were declared.");
        }
        allowedPattern = Pattern.compile(componentScanConfig.allowed());
        deniedPattern = Pattern.compile(componentScanConfig.denied());
        logger.info("@TigrisConfiguration will trace and cache methods into {} and deny {} package.", allowedPattern.pattern(), deniedPattern.pattern());

        samplingConfiguration = configClass.getAnnotation(SamplingConfiguration.class);
        sampling = new Sampling(samplingConfiguration.samplingPercentage(), samplingConfiguration.cycleTimeInSeconds());
        samplingAdaptationExecutor.scheduleWithFixedDelay(
                sampling, 120000, 450000, TimeUnit.MILLISECONDS);

        repository = RepositoryFactory.getRepository(null, tigrisConfiguration.logRepository());

        if (tigrisConfiguration.clearMonitoringDataOnStart()) {
            repository.removeAll();
            logger.debug("Repository starting cleaned.");
        }

        if (coarseMonitoringEnabled) {
            metrics = new ConcurrentHashMap<>();
            lightweightAnalyzerExecutor.scheduleWithFixedDelay(
                    this::run,
                    120000, 450000, TimeUnit.MILLISECONDS);
        }

        try {

            customMonitoringClass = ConfigurationUtils.getAvailableCustomMonitoringClass();
            if (customMonitoringClass != null) {
                customMonitoring = (CustomMonitoring) customMonitoringClass.newInstance();
            }

            customMonitoring.initialize(repository);
        } catch (ConfigurationException e){
            logger.info("AdaptiveCaching not found, disabling...");
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InstantiationException e) {
            e.printStackTrace();
        }
    }

    public boolean isAllowed(ProceedingJoinPoint joinPoint) {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        return
                isEnabled()
                        && !signature.getName().contains("br.ufrgs.inf.prosoft")
                        && allowedPattern.matcher(signature.getMethod().getDeclaringClass().getPackage().getName()).matches()
                        && !deniedPattern.matcher(signature.getMethod().getDeclaringClass().getPackage().getName()).matches();
    }

    public Object process(ProceedingJoinPoint joinPoint) throws Throwable {
        boolean detailedTrace = (!coarseMonitoringEnabled ||
                (coarseMonitoringEnabled && allowedFineGrained
                        .contains(joinPoint.getSignature().toLongString())));

        //generate a hash of the method that will be used as: key to cache and compare if the method is allowed or not
        Key key = new Key(joinPoint);

        boolean customProcess = customMonitoring != null &&
                customMonitoring.beforeExecuting(key);
        if (customProcess) {
            Object returnResult = customMonitoring.returnResult(key);
            if (returnResult != null)
                return returnResult;
        }

        Object[] joinPointArgs = null;
        //method calls can change the args, so it is better to get it at the beginning
        if (detailedTrace)
            joinPointArgs = joinPoint.getArgs();

        String signature = joinPoint.getSignature().toString();

        long startTime = currentTimeMillis();
        Object result = joinPoint.proceed();
        long endTime = currentTimeMillis();

        if (coarseMonitoringEnabled) {
            LightweightMetrics metric = metrics.get(signature);
            if (metric != null)
                metric.incOccurrence();
            else {
                String name = joinPoint.getSignature().toString().split(" ")[1];
                name = name.substring(0, name.indexOf("("));
                metric = new LightweightMetrics(name, joinPoint.getSignature().toLongString());
                metric.incOccurrence();
                metrics.put(signature, metric);
            }
            metric.addTime(startTime, endTime - startTime);
            metric.addReturnSize(result);
        }

        Granularity granularity = new Granularity(samplingConfiguration.granularity(), signature);

        if(sampling.isPerformanceBaselineEnabled()) {
            sampling.addPerformanceBaselineItem(granularity, endTime - startTime);
        }

        //trace only allowed by lightweight metrics
        if (coarseMonitoringEnabled
                && detailedTrace
                && sampling.samplingDecision(granularity)) {
            logger.debug("New trace: " + signature);

            //we do not cache null returns, but we trace them
            //maybe the method can sometimes return null... so there is not verification here
            LogTrace logTrace = new LogTrace();
            logTrace.setStartTime(startTime);
            logTrace.setEndTime(endTime);

            MethodInfo methodInfo = new MethodInfo(joinPoint.getSignature().toLongString(), joinPointArgs, result, key);
            logTrace.setMethodInfo(methodInfo);

            try {
                repository.save(logTrace);
                logger.debug("New trace entry: " + logTrace);
            } catch (Exception e) {
                logger.debug("Couldn't trace " + logTrace.getMethodInfo().getSignature() + " due to: " + e.getMessage());
            }
        }

        if (detailedTrace && customProcess && result != null) {
            customMonitoring.afterExecuting(key, result);
        }

        return result;
    }

    public void turnoff() {
        enabled = false;
    }

    public static boolean isEnabled() {
        return enabled;
    }

    public Repository getRepository() {
        return repository;
    }

    /**
     * The constant allowedFineGrained.
     */
    public static Set<String> allowedFineGrained = new ConcurrentSkipListSet<>();

    @Override
    public void run() {
        try {
            logger.info("Analyzing {} lightweight methods for relevance...", metrics.size());

            StaticMetrics staticMetrics = new StaticMetrics(staticFile);
            DataFiltering dataFiltering = new DataFiltering(staticMetrics, metrics);

            //cleaning selected methods
            allowedFineGrained = new ConcurrentSkipListSet<>();

            //TODO process SET operations dynamically
            // using criteria

            for(String method : metrics.keySet()) {
                LightweightMetrics sc = metrics.get(method);
//                if (dataFiltering.retrictedFilter())
                if (dataFiltering.extendedFilter(sc))
                    allowedFineGrained.add(sc.getLongName());
            }
            logger.info("Selected methods for fine-grained ({}): {}", allowedFineGrained.size(), allowedFineGrained);

            //analyze once
            coarseMonitoringEnabled = false;
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}