An exporter for providing OOM Killer's kills in the Prometheus format. The exporter is compatible with containers runtimes and could be useful to investigate OOM Killer activity inside containers, servers or vms.
import logging
import os
import sys
import time

sys.path.append('./src')
import oom_messages_exporter

log = os.getenv('example_tailer_file')

if __name__ == '__main__':
    logging.basicConfig(format = '%(asctime)s %(message)s', level=20)

def process_a_line(line):
    logging.warning('There is a line: "{}".'.format(line))

tailer = oom_messages_exporter.FileTailer(file=log)

def main():
    while True:
        data = tailer.tail()
        for line in data:
            process_a_line(line)
        time.sleep(0.01)

if __name__ == '__main__':
    main()

examples/filetailer.py
alexeyantropov/oom_messages_exporter
import re
import time
input_string = 'there is a OOM message Ifooph9rahy2uz6shaeveiX4Eiz6koongeex5cao4eRee3ootheit7ahdae1fahsh6choo0aij6aedoo8shuquidahbu5ohHaengadef6iej5Chesohv7heeWei6oosh6koghichookeeN4er5aat3ahQuiepei8peezieXikeehohThoo3Loof8aichauch2dohrooTieshohgh3Aeph2oos4quaiv1choe1Aish8wae1eegoh4tohquiegoo5X'
sub_string = 'oom'
n = 10000000

def with_lower(line):
    return(sub_string in line.lower())

def with_re(line):
    return(re.search(sub_string, line))

re_compiled = re.compile(sub_string)
def with_re_comlite(line):
    return(re_compiled.search(line))

# lower
lower_counter_start = time.perf_counter()
for i in range(n):
    with_lower(input_string)
lower_counter_end = time.perf_counter()

re_counter_start = time.perf_counter()
for i in range(n):
    with_re(input_string)
re_counter_stop = time.perf_counter()

re_com_counter_start = time.perf_counter()
for i in range(n):
    with_re_comlite(input_string)
re_com_counter_stop = time.perf_counter()

print('n: {}'.format(n))
print('Lower: {}'.format(lower_counter_end - lower_counter_start))
print('Re:    {}'.format(re_counter_stop - re_counter_start))
print('Re C:  {}'.format(re_com_counter_stop - re_com_counter_start))
examples/find-substring-perf-test.py
alexeyantropov/oom_messages_exporter
from oom_messages_exporter.exporter import main

import logging

if __name__ == '__main__':
    logging.basicConfig(format = '%(asctime)s %(message)s', level=20)
    main()
src/oom_messages_exporter.py
alexeyantropov/oom_messages_exporter
from .exporter import Exporter
from .cricollector import CriCollector
from .filetailer import FileTailer
src/oom_messages_exporter/__init__.py
alexeyantropov/oom_messages_exporter
import json
import logging
import re
import subprocess
from pathlib import Path
from shutil import which
from typing import TypedDict

class ContainerLabels(TypedDict):
    container_name: str
    pod_name: str
    pod_namespace: str

class PodByPid(TypedDict):
    int: ContainerLabels

class CriCollector:

    """
    The CriCollector class needs the crictl binary.
    According an crictl output the class collects data about namespaces, pods and containers of killed by OOM Killer pids.
    NB: There isn't any Python CRI library :(

    I.e the class transforms a line like this:

    Sep  3 10:00:05 some-kube-node1 kernel: [34268134.409242] \
    oom-kill:constraint=CONSTRAINT_MEMCG,nodemask=(null),cpuset=2cc85334e5030168b240f87215d26ecb97dd99c09237bd0a03d8957dafb33187,\
    mems_allowed=0,oom_memcg=/kubepods/burstable/pod62015ce8-a46f-4b0e-a5a1-3082235c62dd,\
    task_memcg=/kubepods/burstable/pod62015ce8-a46f-4b0e-a5a1-3082235c62dd/2cc85334e5030168b240f87215d26ecb97dd99c09237bd0a03d8957dafb33187,\
    task=framework,pid=3319125,uid=5000

    to:

    -> task_memcg=/kubepods/burstable/poda8d92af7-9288-4612-9741-447b435a8157/4768664dd956fa2530931122cb16f4231368fb4809f7cbd5ca28d583f404517d
    -> 4768664dd956fa2530931122cb16f4231368fb4809f7cbd5ca28d583f404517d
    -> crictl --runtime-endpoint=unix://var/run/containerd/containerd.sock inspect 4768664dd956fa2530931122cb16f4231368fb4809f7cbd5ca28d583f404517d
    -> io.kubernetes.pod.namespace=..., io.kubernetes.pod.name=..., io.kubernetes.container.name=...
    -> {pid: {container_name: '...', pod_name: '...', 'pod_namespace': '...' }}

    """

    def __init__(self, *, test=False) -> None:

        """
        :param test: do not use in production, only for example!
        """

        self.test_binary = './tests/crictl-mock.sh'
        self.test_socket = './tests/cri-test.sock'
        self.collection = dict() # The collection of pids with exta data.
        self.crictl = self.find_crictl(test_binary=test)
        self.socket = self.find_socket(self.crictl, test_socket=test)

    def find_crictl(self, *, test_binary=False, test_path=None) -> str:

        """
        The methods looks for the crictl binary in $PATH and returns abs path to the binary.
        """

        if test_binary:
            logging.warning('Using a test stub for crictl.')
            return(self.test_binary)
        
        tmp = which('crictl', path=test_path)  
        if tmp and len(tmp) > 0:
            logging.warning('The crictl is found: "{}".'.format(tmp))
            return(tmp)

        logging.critical('An exe "crictl" is not found!')
        return(None)
    
    def find_socket(self, crictl: str, *, test_socket=False, run_dir='/var/run/containerd') -> str:

        """
        Looks for a container runtime socket by known socket names in given directory.  
        """

        if test_socket:
            logging.warning('Using a test stub for runtime socket.')
            return(self.test_socket)
        
        if not crictl:
            logging.warning('A path to crictl is not provided.')
            return(None)
        
        known_socks = ['dockershim.sock', 'containerd.sock', 'crio.sock']
        for sock in known_socks:
            p = Path('{}/{}'.format(run_dir, sock)) 
            if p.exists():
                logging.warning('Runtine socket is found: "{}".'.format(str(p)))
                return(str(p))
        
        logging.critical('Runtime socket is not found!')
        return(None)
    
    def cri_inspect(self, id) -> ContainerLabels:

        """
        The crictl binary wrapper. Runs 'crctl inspect' with given container id.
        Returns a dict() with container name, pod and namespace.
        """

        cmd = [
            self.crictl,
            '--runtime-endpoint=unix:/{}'.format(self.socket),
            'inspect',
            id
        ]

        crictl_out = subprocess.run(cmd, capture_output=True, text=True)

        if len(crictl_out.stderr) > 0:
            return(ContainerLabels(container_name='NotFound'))

        try:
            crictl_out_json = json.loads(crictl_out.stdout)
        except:
            return(ContainerLabels(container_name='NotFound'))

        if 'status' in crictl_out_json and 'labels' in crictl_out_json['status']:
            ret = ContainerLabels()
            labels = crictl_out_json['status']['labels']

            if 'io.kubernetes.container.name' in labels:
                ret['container_name'] = labels['io.kubernetes.container.name']
            if 'io.kubernetes.pod.name' in labels:
                ret['pod_name'] = labels['io.kubernetes.pod.name']
            if 'io.kubernetes.pod.namespace' in labels:
                ret['pod_namespace'] = labels['io.kubernetes.pod.namespace']

            return(ret)
        
        # Nothing is found, return an empty data set.
        return(ContainerLabels(container_name='NotFound'))
    
    def get_pid_from_oom_line(self, line: str) -> int:

        """
        A simple method extracts pid from oom killer line from dmegs or messages log by given line.
        """

        regex = r'pid=(\d*),'
        tmp = re.search(regex, line, flags=re.IGNORECASE)
        if tmp and tmp.group(1).isdigit():
            logging.warning('Got pid: "{}"'.format(tmp.group(1)))
            return(int(tmp.group(1)))
        
        return(-1)
    
    def get_containerID_from_oom_line(self, line: str) -> str:

        """
        A simple method extracts container id from oom killer line from dmegs or messages log by given line.
        """

        # ...,task_memcg=/kubepods/burstable/poda8d92af7-9288-4612-9741-447b435a8157/57953e6c546fa43d1ebadf04b16d4e324378ad03d704dadf2cd41a9481eae3f8,...
        regex = r'task_memcg=([\/a-zA-Z0-9\-]+)'
        tmp = re.search(regex, line, flags=re.IGNORECASE)

        if tmp and tmp.group(1):
            ct_id =  tmp.group(1).split('/')[-1]
            logging.warning('Got ct id: "{}"'.format(ct_id))
            return(ct_id)
        
        return(None)
            
    def add(self, line: str) -> int:

        """
        The main method adds info about pid in the collection by given line with 'oom-kill' signature.
        """

        if not (self.crictl and self.socket):
            logging.critical('Vars self.crictl: "{}" or self.socket: "{}" are not set!'.format(self.crictl, self.socket))
            return(-1)
        if not line:
            return(-1)
        
        pid = self.get_pid_from_oom_line(line)
        container_id = self.get_containerID_from_oom_line(line)

        if pid < 0 or not container_id:
            return(-1)
        
        tmp = self.cri_inspect(container_id)
        if len(tmp) > 0:
            self.collection[pid] = tmp
            return(pid)
        
        return(-1)
    
    def get(self, pid: int) -> ContainerLabels:

        """
        The method returns extra info about a given pid from the collection.
        """

        if pid in self.collection:
            return(self.collection[pid])
        
        return(None)
src/oom_messages_exporter/cricollector.py
alexeyantropov/oom_messages_exporter
import logging
import os
import time
import re

from .filetailer import FileTailer
from .cricollector import CriCollector

from prometheus_client import Gauge, start_http_server

class Exporter:

    """
    The class provides an exporter for collecting info about OOM Killer works from system logs.
    """

    def __init__(self, *, log: str, interval=1, test_enable='no'):

        """
        :param log: by default if's /var/log/messages.
        :param interval: the log pooling interval.
        :param test_enable: do not use it.
        """

        self.log = log
        self.interval = interval
        self.tailer = self.__tailer_setup()
        if self.tailer is None: # You can't init the class if log does not exist!
            raise FileNotFoundError(log)
        self.metrics_name_for_pid = 'oom_messages_exporter_last_killed_pid'
        self.metrics_name_for_time = 'oom_messages_exporter_last_killed_time'
        self.oom_sign_cgroup = 'Killed process'
        self.container_sign = '] oom-kill:'
        self.__metrics_setup()
        if test_enable.lower() == 'no':
            self.pid_collection = CriCollector()
        else:
            self.pid_collection = CriCollector(test=True)

    def __metrics_setup(self):

        """
        Initialization of prometheus metrics.
        """

        labels = ['command', 'namespace', 'pod', 'container']
        self.last_killed_pid = Gauge(self.metrics_name_for_pid,
            'The last pid killed by OOM killer',
            labels)
        self.last_killed_time = Gauge(self.metrics_name_for_time,
            'The last time when a pid is killed by OOM killer',
            labels)
        
    def __tailer_setup(self) -> FileTailer:

        """
        Initialization of the file tailer.
        """

        try:
            return(FileTailer(file=self.log))
        except Exception as err:
            logging.critical('Can not setup a tailer on the "{}" log!'.format(self.log))
            return(None)
        
    def __string_has_oom(self, line: str) -> bool:

        """
        Checks a given line and returns 'True' if the line seems like a message about OOM Killer. 
        """

        return(self.oom_sign_cgroup.lower() in line.lower())
    
    def __string_has_container(self, line: str) -> bool:

        """
        Checks a given line and returns 'True' if the line has an information about a container runtime.
        """
        
        return(self.container_sign.lower() in line.lower())
    
    def __extract_pid(self, line: str, sign: str) -> int:

        """
        Gets a pid from a given line if OOM has killed a process in the line.
        """

        regexp = r'{} (\d*)'.format(sign)
        m = re.search(regexp, line, flags=re.IGNORECASE)
        if m and m.group(1).isdigit():
            logging.warning('Got pid: "{}"'.format(m.group(1)))
            return(int(m.group(1)))
        return(-1)
    
    def __extract_proc_name(self, line: str, sign: str) -> str:
        """
        Gets a process name from a given line if OOM has killed a process in the line.
        """
        regexp = r'{} \d* \((.*)\)\,?'.format(sign)
        m = re.search(regexp, line, flags=re.IGNORECASE)
        if m and m.group(1):
            logging.warning('Got proc name: "{}"'.format(m.group(1)))
            return(m.group(1))
        return('Could not get a proc name from a line.')
    
    def __extract_cri_data(self, pid: int) -> list:

        """
        The method returns extra labels by a given pid if the killed pid was in a container.
        """

        pod_namespace, pod_name, container_name = None, None, None
        container_data = self.pid_collection.get(pid)
        
        if container_data:
            if 'pod_namespace' in container_data:
                pod_namespace = container_data['pod_namespace']
            if 'pod_name' in container_data:
                pod_name = container_data['pod_name']
            if 'container_name' in container_data:
                container_name = container_data['container_name']

        return([pod_namespace, pod_name, container_name])

    def __process_data(self, line: str) -> str:

        """
        The method processes each given line and tries to recognize needed data about OOM Killer.
        """
        
        # Data about container should saved into additional storage for future use.
        if self.__string_has_container(line):

            logging.warning('Have got container data, transmit the data to pid collector')
            tmp = self.pid_collection.add(line)

            if tmp > 0:
                logging.warning('Container is added for a pid: "{}"'.format(tmp))
                return('has container: is added')
            
            return('has container: is not added')

        # Counters would be incremented if OOM Killer killed a process.
        # Also the code below tries to add extra data about the killed process if the process worked in a container runtime.
        if self.__string_has_oom(line):
            logging.warning('Got OOM Killer in a line')

            pid = self.__extract_pid(line, self.oom_sign_cgroup)
            labels = [self.__extract_proc_name(line, self.oom_sign_cgroup)] + self.__extract_cri_data(pid)
            self.last_killed_pid.labels(*labels).set(pid)
            self.last_killed_time.labels(*labels).set_to_current_time()

            return('has oom')

        # If nothing interesting were found the code would write a dummy string.
        n = 50
        logging.debug('Got string w/o OOM, the first {} symbols: "{}"'.format(n, line[0:n]))

        return('has not any')

    def run_metrics_loop(self, *, test_runs=-1, test_data='') -> bool:
        
        """
        The main endless loop. The loop calls tail() method from the file tailer and
        will process lines if the lines appear in the log.
        """

        i = 0
        while True:

            data = self.tailer.tail()

            # The following block only for testing!
            if test_runs > 0 and i > test_runs:
                break
            if test_runs > 0:
                i += 1
            if test_data: 
                data = test_data
            
            if len(data) > 0:
                logging.debug('Data occurred in log, amount of lines: "{}"'.format(len(data)))
            for line in data:
                self.__process_data(line)
                
            time.sleep(self.interval)

        return(True)
           
def main(): # pragma: no cover
    # Config
    test_enable = os.getenv('oom_messages_exporter_TEST', 'no')
    debug_enable = os.getenv('oom_messages_exporter_DEBUG', 'no')
    exporter_port = int(os.getenv('oom_messages_exporter_PORT', 9001))
    messages_log = os.getenv('oom_messages_exporter_MESSAGES_LOG', '/var/log/messages')
    poll_interval = int(os.getenv('oom_messages_exporter_POLL_INTERVAL', 1))
    
    # Set debug logging
    if debug_enable != 'no':
        logging.getLogger().setLevel(logging.DEBUG)
    logging.debug('debug is enable')

    # There is the metrics
    start_http_server(int(exporter_port))

    # The exporter setup
    ex = Exporter(log=messages_log, interval=poll_interval, test_enable=test_enable)
    logging.warning('oom_messages_exporter started! port: {}.'.format(exporter_port))

    # Run it!
    ex.run_metrics_loop()

src/oom_messages_exporter/exporter.py
alexeyantropov/oom_messages_exporter
import logging
import os
from typing import TypedDict

def log_exception(where, err):
    logging.critical('An error occurred in "{}", exception:\n---\n{}\n---'.format(where, err))

class FileStat(TypedDict):
    inode: int
    size: int

class FileTailer:
    """
    The class works like the 'tail -F <file>' command and returns a set of lines if they've appeared in a given log file since last call.
    You need only initialize the class and use the tail() method, like this example:

    ft = FileTailer(file='/var/log/messages')
    while True:
        data = ft.tail()
        if len(data) > 0:
            process...(data)
        sleep(how many that you want)
    """
    def __init__(self, *, file) -> None:

        """
        :param file: a log that will be watched
        """

        self.file_name = file
        self.file_stats = FileStat # Data in this variable is useful for checking the given file rotate. 
        self.fd = None

        if self.__open_file():
            pass
        else:
            raise FileNotFoundError(file)
        
    def __open_file(self, pos=-1) -> (bool, str):

        """
        The method opens a file and moves a pointer to the last line.
        Also the method stores info about the opened file thought a call of the '__update_file_stat' method.
        """

        self.__update_file_stat()

        if not os.path.exists(self.file_name):
            logging.error('File "{}" does not exist.'.format(self.file_name))
            return(False)
        
        try:
            fd = open(self.file_name)
        except Exception as err:
            log_exception('open({})'.format(self.file_name), err)
            return(False)
        
        # If needed positions isn't set -> move a pointer in a fd to the end.
        self.fd = fd
        if pos < 0:
            self.fd.seek(self.file_stats['size'])

        return(True)

    def __get_stat(self) -> FileStat:

        """
        The method returns info about the give file. Size, inode, etc.
        This data is used for check a rotation of the file.
        """

        ret = {'inode': -1, 'size': -1}

        try:
            stat = os.stat(self.file_name)
            ret['inode'] = stat.st_ino
            ret['size'] = stat.st_size
        except Exception as err:
            log_exception('__get_stat', err)

        return(ret)
    
    def __update_file_stat(self) -> None:

        """
        This methods stores and updates info about the given file in the 'file_stats' variable.
        """

        self.file_stats = self.__get_stat()
    
    def __file_is_rotated(self) -> bool:

        """
        This method checks a rotation of the given log and returns 'True' if the log has been rotated.
        Rotations by move and by truncate are supported.
        """

        prev_inode = self.file_stats['inode']
        self.__update_file_stat()

        if self.file_stats['inode'] != prev_inode:
            logging.warning('The "{}" file is rotated, inode: {} -> {}'.format(self.file_name, prev_inode, self.file_stats['inode']))
            return(True)
        
        if self.file_stats['size'] < self.fd.tell():
            logging.warning('The "{}" file is truncated, size: {} -> {}'.format(self.file_name, self.fd.tell(), self.file_stats['size']))
            return(True)
        
        return(False)
    
    def tail(self) -> list[str]:

        """
        The main method of the class. Method returns a list of lines since the last call of the method.
        You have to call the method from time to time (few seconds). When the method will return a non empty list
        it's a mark that you can do smth with the returned lines.
        """

        ret = list()

        if self.file_stats['inode'] == -1 and self.__open_file(0) == False:
            logging.critical('Can not open the "{}" file.'.format(self.file_name))
            return(list())

        line = self.fd.readline()
        while line:
            ret.append(line.strip())
            line = self.fd.readline()

        if self.__file_is_rotated():
            if self.__open_file(0):
                logging.warning('The "{}" file is reopened, inode: {}.'.format(self.file_name, self.file_stats['inode']))
            else:
                logging.critical('The "{}" file is NOT reopened, will try again next time'.format(self.file_name))

        return(ret)
src/oom_messages_exporter/filetailer.py
alexeyantropov/oom_messages_exporter
'''
A little convention about test names: test_(1)___(2), where:
1 - Name of a functionality or a method;
2 - A clarification or a detail describes the test.
'''

import sys

sys.path.append('./src')
import oom_messages_exporter

line_with_oom_pid = 3319125
line_with_oom_container_id = '2cc85334e5030168b240f87215d26ecb97dd99c09237bd0a03d8957dafb33187'
line_with_oom_template = 'Sep  3 10:00:05 some-kube-node1 kernel: [34268134.409242] oom-kill:constraint=CONSTRAINT_MEMCG,nodemask=(null),cpuset=2cc85334e5030168b240f87215d26ecb97dd99c09237bd0a03d8957dafb33187,mems_allowed=0,oom_memcg=/kubepods/burstable/pod62015ce8-a46f-4b0e-a5a1-3082235c62dd,task_memcg=/kubepods/burstable/pod62015ce8-a46f-4b0e-a5a1-3082235c62dd/{},task=framework,pid={},uid=5000'
line_with_oom = line_with_oom_template.format(line_with_oom_container_id, line_with_oom_pid)
line_with_oom_not_existen_container = line_with_oom_template.format('dummy', line_with_oom_pid)

cc = oom_messages_exporter.CriCollector()

def test_find_crictl___test_binary():
    assert cc.test_binary == './tests/crictl-mock.sh'
    assert cc.find_crictl(test_binary=True) == cc.test_binary

def test_find_crictl___test_patch():
    assert cc.find_crictl(test_path='./tests') == './tests/crictl'

def test_find_socket___test_socket():
    assert cc.test_socket == './tests/cri-test.sock'
    assert cc.find_socket(crictl='dummy', test_socket=True) == cc.test_socket

def test_find_socket___wo_crictl():
    assert cc.find_socket(crictl=None) is None

def test_find_socket___not_found():
    assert cc.find_socket(crictl='dummy') is None

def test_find_socket___test_dir():
    assert cc.find_socket(crictl='dummy', run_dir='./tests') == 'tests/containerd.sock'

def test_get_pid_from_oom_line___Ok():
    assert cc.get_pid_from_oom_line(line_with_oom) == line_with_oom_pid

def test_get_pid_from_oom_line___Fail():
    assert cc.get_pid_from_oom_line('blahblahblah') == -1

def test_get_containerID_from_oom_line___Well():
    assert cc.get_containerID_from_oom_line(line_with_oom) == line_with_oom_container_id

def test_get_containerID_from_oom_line___Fail():
    assert cc.get_containerID_from_oom_line('blahblahblah') is None

def test_add___no_crictl():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = None, cc_mocked.test_socket
    assert cc_mocked.add(line_with_oom) == -1

def test_add___no_socket():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = cc_mocked.test_binary, None
    assert cc_mocked.add(line_with_oom) == -1

def test_add___no_line():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = cc_mocked.test_binary, cc_mocked.test_socket
    assert cc_mocked.add('') == -1

def test_add___Wrong_Line():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = cc_mocked.test_binary, cc_mocked.test_socket
    assert cc_mocked.add('dummy') == -1

def test_add___crictl_mock_fail():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = cc_mocked.test_binary, cc_mocked.test_socket
    assert cc_mocked.add(line_with_oom_not_existen_container) == line_with_oom_pid
    assert cc_mocked.get(line_with_oom_pid)['container_name'] == 'NotFound'

def test_add___crictl_mock_no_json():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = './tests/crictl-mock_no_json.sh', cc_mocked.test_socket
    assert cc_mocked.add(line_with_oom_not_existen_container) == line_with_oom_pid
    assert cc_mocked.get(line_with_oom_pid)['container_name'] == 'NotFound'

def test_add___crictl_mock_empty_json():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = './tests/crictl-mock_empty_json.sh', cc_mocked.test_socket
    assert cc_mocked.add(line_with_oom_not_existen_container) == line_with_oom_pid
    assert cc_mocked.get(line_with_oom_pid)['container_name'] == 'NotFound'
    
def test_add___Well():
    cc_mocked = oom_messages_exporter.CriCollector()
    cc_mocked.crictl, cc_mocked.socket = cc_mocked.test_binary, cc_mocked.test_socket
    assert cc_mocked.add(line_with_oom) == line_with_oom_pid

def test_get___Well():
    cc_mocked = oom_messages_exporter.CriCollector()
    pid, name, data = 123, 'test', dict()
    data['container_name'] = name
    cc_mocked.crictl, cc_mocked.socket = cc_mocked.test_binary, cc_mocked.test_socket
    cc_mocked.collection[pid] = data
    assert cc_mocked.get(pid)['container_name'] == name

def test_get___Wrong():
    cc_mocked = oom_messages_exporter.CriCollector()
    assert cc_mocked.get(1234) is None
tests/cricollector.py
alexeyantropov/oom_messages_exporter
'''
A little convention about test names: test_(1)___(2), where:
1 - Name of a functionality or a method;
2 - A clarification or a detail describes the test.
'''

import sys
import os
import pytest

sys.path.append('./src')
import oom_messages_exporter

file_d_lines = os.getenv('file_d_lines')
non_existent_file = '/foo/bar/baz/txt'
string_with_oom_pid = 1234
string_with_oom_proc_name = 'node /opt/www/p'
string_with_oom = 'Memory cgroup out of memory: Kill process {} ({}) score 331 or sacrifice chil'.format(string_with_oom_pid, string_with_oom_proc_name)
string_with_oom = 'Sep  4 00:05:58 other-server kernel: [38469674.213429] Killed process {} ({}), UID 80, total-vm:1203112kB, anon-rss:329556kB, file-rss:15708kB, shmem-rss:0kB'.format(string_with_oom_pid, string_with_oom_proc_name)
string_wo_oom = 'just a line'

cri_line_with_oom_pid = string_with_oom_pid
cri_line_with_oom_container_id = '2cc85334e5030168b240f87215d26ecb97dd99c09237bd0a03d8957dafb33187'
cri_line_with_oom = 'Sep  3 10:00:05 some-kube-node1 kernel: [34268134.409242] oom-kill:constraint=CONSTRAINT_MEMCG,nodemask=(null),cpuset=2cc85334e5030168b240f87215d26ecb97dd99c09237bd0a03d8957dafb33187,mems_allowed=0,oom_memcg=/kubepods/burstable/pod62015ce8-a46f-4b0e-a5a1-3082235c62dd,task_memcg=/kubepods/burstable/pod62015ce8-a46f-4b0e-a5a1-3082235c62dd/{},task=framework,pid={},uid=5000'.format(cri_line_with_oom_container_id, cri_line_with_oom_pid)

def test_init___Fail_Not_exist():
    with pytest.raises(Exception):
        ex_ne = oom_messages_exporter.Exporter(log=non_existent_file)

ex = oom_messages_exporter.Exporter(log=file_d_lines)
ex.pid_collection.crictl = ex.pid_collection.test_binary
ex.pid_collection.socket = ex.pid_collection.test_socket

def test_init___Ok():
    assert ex is not None

def test_string_has_oom___is_True():
    assert ex._Exporter__string_has_oom(string_with_oom) == True

def test_string_has_oom___is_not_True():
    assert ex._Exporter__string_has_oom(string_wo_oom) == False

def test_extract_pid___Ok():
    oom_sign_cgroup = ex.oom_sign_cgroup
    assert ex._Exporter__extract_pid(string_with_oom, oom_sign_cgroup) == string_with_oom_pid

def test_extract_pid___Fail():
    oom_sign_cgroup = ex.oom_sign_cgroup
    assert ex._Exporter__extract_pid(string_wo_oom, oom_sign_cgroup) == -1

def test_extract_proc_name___Ok():
    oom_sign_cgroup = ex.oom_sign_cgroup
    assert ex._Exporter__extract_proc_name(string_with_oom, oom_sign_cgroup) == string_with_oom_proc_name

def test_extract_proc_name___Fail():
    oom_sign_cgroup = ex.oom_sign_cgroup
    assert ex._Exporter__extract_proc_name(string_wo_oom, oom_sign_cgroup) == 'Could not get a proc name from a line.'

def test_process_data___OOM_Sign_is_present():
    ex._Exporter__process_data(cri_line_with_oom)
    assert ex._Exporter__process_data(string_with_oom) == 'has oom'

def test_process_data___OOM_Sign_not_is_present():
    assert ex._Exporter__process_data(string_wo_oom) == 'has not any'

def test_process_data___Cri_data_is_present():
    assert ex._Exporter__process_data(cri_line_with_oom) == 'has container: is added'

def test_run_metrics_loop():
    assert ex.run_metrics_loop(test_runs=1, test_data=string_with_oom) == True

tests/exporter.py
alexeyantropov/oom_messages_exporter
'''
A little convention about test names: test_(1)___(2), where:
1 - Name of a functionality or a method;
2 - A clarification or a detail describes the test.
'''

import sys
import os
import pytest

sys.path.append('./src')
import oom_messages_exporter

file_d_lines = os.getenv('file_d_lines')
file_zero_mode = os.getenv('file_zero_mode')
non_existent_file = '/foo/bar/baz/txt'
rotated_file = '/tmp/rotated.file'

ft = oom_messages_exporter.FileTailer(file=file_d_lines)

def recreate_a_dummy_file(f):
    if os.path.exists(f):
        os.remove(f)
    fd = open(f, 'w')
    fd.write('data\n')
    fd.close()

def add_strings_to_file(f, count):
    fd = open(f, 'a')
    for i in range(count):
        fd.write('data: {}\n'.format(i))
    fd.close()

def truncate_a_file(f):
    fd = open(f, 'a')
    fd.truncate(0)
    fd.close

def test_open___fail_1():
    with pytest.raises(Exception):
        ft_ne = oom_messages_exporter.FileTailer(file=non_existent_file)

def test_open___fail_2():
    with pytest.raises(Exception):
        oom_messages_exporter.FileTailer(file=file_zero_mode)

def test_open___well():
    assert ft._FileTailer__open_file() is True
    
def test_get_stat___fail():
    ft_gs = oom_messages_exporter.FileTailer(file=file_d_lines)
    ft_gs.file_name = non_existent_file
    assert ft_gs._FileTailer__get_stat() == {'inode': -1, 'size': -1}

def test_file_is_rotated___False():
    recreate_a_dummy_file(rotated_file)
    ft_rotated = oom_messages_exporter.FileTailer(file=rotated_file)
    add_strings_to_file(rotated_file, 2)
    assert ft_rotated._FileTailer__file_is_rotated() == False

def test_file_is_rotated___True():
    ft_rotated = oom_messages_exporter.FileTailer(file=rotated_file)
    recreate_a_dummy_file(rotated_file)
    assert ft_rotated._FileTailer__file_is_rotated() == True

def test_file_is_rotated___Truncated():
    recreate_a_dummy_file(rotated_file)
    add_strings_to_file(rotated_file, 2)
    ft_rotated = oom_messages_exporter.FileTailer(file=rotated_file)
    truncate_a_file(rotated_file)
    assert ft_rotated._FileTailer__file_is_rotated() == True

def test_tail___Ok_Simple():
    ft_tail = oom_messages_exporter.FileTailer(file=file_d_lines)
    assert len(ft_tail.tail()) == 0

def test_tail___Ok_NewLines():
    recreate_a_dummy_file(rotated_file)
    ft_tail = oom_messages_exporter.FileTailer(file=rotated_file)
    ft_tail.tail()
    how_to_add = [3, 33]
    for amount_of_lines in how_to_add:
        add_strings_to_file(rotated_file, amount_of_lines)
        assert len(ft_tail.tail()) == amount_of_lines

def test_tail___Ok_Rotated():
    recreate_a_dummy_file(rotated_file)
    add_strings_to_file(rotated_file, 2)
    ft_tail = oom_messages_exporter.FileTailer(file=rotated_file)
    ft_tail.tail()
    recreate_a_dummy_file(rotated_file)
    assert len(ft_tail.tail()) == 0

def test_tail___Deleted():
    recreate_a_dummy_file(rotated_file)
    ft_tail = oom_messages_exporter.FileTailer(file=rotated_file)
    ft_tail.tail()
    add_strings_to_file(rotated_file, 1)
    os.remove(rotated_file)
    assert len(ft_tail.tail()) == 1
    assert len(ft_tail.tail()) == 0

tests/filetailer.py
alexeyantropov/oom_messages_exporter