[Jenkins]Jenkins

Code Coverage

Cobertura Coverage Report > osm_lcm >

lcm_hc.py

Trend

File Coverage summary

NameClassesLinesConditionals
lcm_hc.py
100%
1/1
58%
21/36
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
lcm_hc.py
58%
21/36
N/A

Source

osm_lcm/lcm_hc.py

1

#!/usr/bin/python3




2

# -*- coding: utf-8 -*-




3






4

##




5

# Copyright 2018 Telefonica S.A.




6

#




7

# Licensed under the Apache License, Version 2.0 (the "License"); you may




8

# not use this file except in compliance with the License. You may obtain




9

# a copy of the License at




10

#




11

#         http://www.apache.org/licenses/LICENSE-2.0




12

#




13

# Unless required by applicable law or agreed to in writing, software




14

# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT




15

# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the




16

# License for the specific language governing permissions and limitations




17

# under the License.




18

##




19






20
1
from time import time, sleep




21
1
from sys import stderr




22






23
1
from osm_lcm.lcm_utils import LcmException




24
1
import yaml




25






26

""" This module is used for health check. A file called time_last_ping is used




27

This contains the last time where something is received from kafka




28

"""




29






30






31
1
def get_health_check_file(config_file=None):




32
1
    try:




33
1
        health_check_file = "/app/storage/time_last_ping"




34
1
        if not config_file:




35
1
            return health_check_file




36

        # If config_input is dictionary




37
1
        if isinstance(config_file, dict) and config_file.get("storage"):




38
1
            health_check_file = config_file["storage"]["path"] + "/time_last_ping"




39

        # If config_input is file




40
1
        elif isinstance(config_file, str):




41
1
            with open(config_file) as f:




42

                # read file as yaml format




43
1
                conf = yaml.safe_load(f)




44

                # Ensure all sections are not empty




45
1
                if conf.get("storage"):




46
1
                    health_check_file = conf["storage"]["path"] + "/time_last_ping"




47






48
1
        return health_check_file




49
1
    except (IOError, FileNotFoundError, TypeError, AttributeError, KeyError) as error:




50
1
        raise LcmException(




51

            f"Error occured while getting the health check file location: {error}"




52

        )




53






54






55
1
def health_check(config_file=None, ping_interval_pace=120):




56
0
    health_check_file = get_health_check_file(config_file)




57
0
    retry = 2




58
0
    while retry:




59
0
        retry -= 1




60
0
        try:




61
0
            with open(health_check_file, "r") as f:




62
0
                last_received_ping = f.read()




63






64
0
            if (




65

                time() - float(last_received_ping) < 2 * ping_interval_pace




66

            ):  # allow one ping not received every two




67
0
                exit(0)




68
0
        except Exception as e:




69
0
            print(e, file=stderr)




70
0
        if retry:




71
0
            sleep(6)




72
0
    exit(1)




73






74






75
1
if __name__ == "__main__":




76
0
    health_check()