<commit_before>package com.github.bingoohuang.blackcat.agent.collectors;

import com.alibaba.fastjson.util.IOUtils;
import com.github.bingoohuang.blackcat.sdk.netty.BlackcatReqSender;
import com.github.bingoohuang.blackcat.sdk.protobuf.BlackcatMsg;
import com.github.bingoohuang.blackcat.sdk.utils.Blackcats;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.EvictingQueue;
import com.google.common.collect.Lists;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.val;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

import static com.github.bingoohuang.blackcat.sdk.protobuf.BlackcatMsg.BlackcatReqHead.ReqType.BlackcatLogException;

@AllArgsConstructor
public class BlactcatLogExceptionCollector {
    private final BlackcatReqSender sender;
    private final String logFiles;
    private final long rotateSeconds;

    @SneakyThrows
    public void start() {
        val processBeans = parseLogConfigs();
        if (processBeans.isEmpty()) return;

        val executorService = Executors.newSingleThreadScheduledExecutor();
        executorService.scheduleWithFixedDelay(() -> {
            for (val processBean : processBeans) {
                processBean.rotateCheck();
            }
        }, rotateSeconds, rotateSeconds, TimeUnit.SECONDS);
    }

    private List<ProcessBean> parseLogConfigs() {
        val loggers = Splitter.on(',').omitEmptyStrings().trimResults().splitToList(logFiles);
        val processBeans = Lists.<ProcessBean>newArrayListWithCapacity(loggers.size());
        for (val logFile : loggers) {
            val loggerAndFile = logFile.split(":");
            val commands = new String[]{"/bin/bash", "-c", "tail -F " + loggerAndFile[1], "&"};

            processBeans.add(new ProcessBean(sender, commands, loggerAndFile[0]));
        }
        return processBeans;
    }


    public static class LineReader {
        private final BufferedReader reader;
        private StringBuilder lineStringBuilder = new StringBuilder();

        public LineReader(BufferedReader reader) {
            this.reader = reader;
        }

        private String readLine() throws IOException {
            while (reader.ready()) {
                val chr = reader.read();
                if (chr == -1) break;

                lineStringBuilder.append((char) chr);
                if (chr == '\n') {
                    val line = lineStringBuilder.toString();
                    lineStringBuilder.setLength(0);
                    return line;
                }
            }

            return null;
        }
    }

    interface Consts {
        Pattern NORMAL_LINE_PATTERN = Pattern.compile("^(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d{3}) .*", Pattern.DOTALL); // 2017-11-20 03:58:48.167
        Pattern EXCEPTION_PATTERN = Pattern.compile("\\b\\S+Exception:"); // java.io.InvalidClassException:
        // tenantCode[1704421450] tenantId[94c63603-e487-4cb3-bc98-6129ec616722]
        Pattern TCODE_PATTERN = Pattern.compile("tenantCode\\[(\\d+)\\]");
        Pattern TID_PATTERN = Pattern.compile("tenantId\\[(.*?)\\]");

        Joiner JOINER = Joiner.on("");
    }

    @Data @AllArgsConstructor
    private static class ProcessBean {
        private final BlackcatReqSender sender;
        private final String[] commands;
        private final String logger;

        private Process process;
        private BufferedReader bufferedReader;
        private LineReader lineReader;

        private EvictingQueue<Object> evictingQueue = EvictingQueue.create(10);
        private List<String> exceptionStack = new ArrayList<>();

        public ProcessBean(BlackcatReqSender sender, String[] commands, String logger) {
            this.sender = sender;
            this.commands = commands;
            this.logger = logger;

            createLineReader(commands);
        }

        public void reset() {
            process.destroy();
            IOUtils.close(bufferedReader);

            evictingQueue.clear();
            exceptionStack.clear();
            createLineReader(commands);
        }

        @SneakyThrows
        private void createLineReader(String[] commands) {
            process = Runtime.getRuntime().exec(commands);
            bufferedReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
            lineReader = new LineReader(bufferedReader);
        }

        public void rotateCheck() {
            String lastNormalLine = null;
            try {
                while (true) {
                    val line = lineReader.readLine();
                    if (line == null) break;

                    if (Consts.NORMAL_LINE_PATTERN.matcher(line).matches()) {
                        detectException(lastNormalLine);
                        evictingQueue.add(line);
                        lastNormalLine = line;
                    } else {
                        exceptionStack.add(line);
                    }
                }
            } catch (IOException ex) {
                reset();
            }

            detectException(lastNormalLine);
        }<commit_msg>add process alive detection<commit_after>package com.github.bingoohuang.blackcat.agent.collectors;

import com.alibaba.fastjson.util.IOUtils;
import com.github.bingoohuang.blackcat.sdk.netty.BlackcatReqSender;
import com.github.bingoohuang.blackcat.sdk.protobuf.BlackcatMsg;
import com.github.bingoohuang.blackcat.sdk.utils.Blackcats;
import com.github.bingoohuang.blackcat.sdk.utils.ProcessExecutor;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.EvictingQueue;
import com.google.common.collect.Lists;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.val;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

import static com.github.bingoohuang.blackcat.sdk.protobuf.BlackcatMsg.BlackcatReqHead.ReqType.BlackcatLogException;

@AllArgsConstructor
public class BlactcatLogExceptionCollector {
    private final BlackcatReqSender sender;
    private final String logFiles;
    private final long rotateSeconds;

    @SneakyThrows
    public void start() {
        val processBeans = parseLogConfigs();
        if (processBeans.isEmpty()) return;

        val executorService = Executors.newSingleThreadScheduledExecutor();
        executorService.scheduleWithFixedDelay(() -> {
            for (val processBean : processBeans) {
                processBean.rotateCheck();
            }
        }, rotateSeconds, rotateSeconds, TimeUnit.SECONDS);
    }

    private List<ProcessBean> parseLogConfigs() {
        val loggers = Splitter.on(',').omitEmptyStrings().trimResults().splitToList(logFiles);
        val processBeans = Lists.<ProcessBean>newArrayListWithCapacity(loggers.size());
        for (val logFile : loggers) {
            val loggerAndFile = logFile.split(":");
            val commands = new String[]{"/bin/bash", "-c", "tail -F " + loggerAndFile[1], "&"};

            processBeans.add(new ProcessBean(sender, commands, loggerAndFile[0]));
        }
        return processBeans;
    }


    public static class LineReader {
        private final BufferedReader reader;
        private StringBuilder lineStringBuilder = new StringBuilder();

        public LineReader(BufferedReader reader) {
            this.reader = reader;
        }

        private String readLine() throws IOException {
            while (reader.ready()) {
                val chr = reader.read();
                if (chr == -1) break;

                lineStringBuilder.append((char) chr);
                if (chr == '\n') {
                    val line = lineStringBuilder.toString();
                    lineStringBuilder.setLength(0);
                    return line;
                }
            }

            return null;
        }
    }

    interface Consts {
        Pattern NORMAL_LINE_PATTERN = Pattern.compile("^(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d{3}) .*", Pattern.DOTALL); // 2017-11-20 03:58:48.167
        Pattern EXCEPTION_PATTERN = Pattern.compile("\\b\\S+Exception:"); // java.io.InvalidClassException:
        // tenantCode[1704421450] tenantId[94c63603-e487-4cb3-bc98-6129ec616722]
        Pattern TCODE_PATTERN = Pattern.compile("tenantCode\\[(\\d+)\\]");
        Pattern TID_PATTERN = Pattern.compile("tenantId\\[(.*?)\\]");

        Joiner JOINER = Joiner.on("");
    }

    @Data @AllArgsConstructor
    private static class ProcessBean {
        private final BlackcatReqSender sender;
        private final String[] commands;
        private final String logger;

        private Process process;
        private BufferedReader bufferedReader;
        private LineReader lineReader;

        private EvictingQueue<Object> evictingQueue = EvictingQueue.create(10);
        private List<String> exceptionStack = new ArrayList<>();

        public ProcessBean(BlackcatReqSender sender, String[] commands, String logger) {
            this.sender = sender;
            this.commands = commands;
            this.logger = logger;

            createLineReader(commands);
        }

        public void reset() {
            process.destroy();
            IOUtils.close(bufferedReader);

            evictingQueue.clear();
            exceptionStack.clear();
            createLineReader(commands);
        }

        @SneakyThrows
        private void createLineReader(String[] commands) {
            process = Runtime.getRuntime().exec(commands);
            bufferedReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
            lineReader = new LineReader(bufferedReader);
        }

        public void rotateCheck() {
            if (!ProcessExecutor.isAlive(process)) {
                reset();
            }

            String lastNormalLine = null;
            try {
                while (true) {
                    val line = lineReader.readLine();
                    if (line == null) break;

                    if (Consts.NORMAL_LINE_PATTERN.matcher(line).matches()) {
                        detectException(lastNormalLine);
                        evictingQueue.add(line);
                        lastNormalLine = line;
                    } else {
                        exceptionStack.add(line);
                    }
                }
            } catch (Exception ex) {
                reset();
            }

            detectException(lastNormalLine);
        }
{'fraction_non_alphanumeric': 0.1070154577883472, 'fraction_numerical': 0.01170767401445166, 'mean_word_length': 2.266806095010457, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6894430', 'n_tokens_mistral': 3073, 'n_tokens_neox': 2897, 'n_words': 574}