Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def read_telemetry_entries():
"""
Read and return the telemetry entries.
:return: List[(str, dict)] returns a list of tuples representing the input
to the track_event command for the application insights telemetry client.
Each tuple contains the following:
(str) the command_name being called, without parameters. For example, 'node show'
(dict) See return from get_telemetry_input_as_dict()
"""
return_tuples = []
# Mode r starts at the beginning of the file
with portalocker.Lock(TELEMETRY_FILE_PATH, timeout=1, fail_when_locked=True, mode='r') as telemetry_file: # pylint: disable=line-too-long
all_lines = telemetry_file.readlines()
# Parse the lines. The lines have format {0}, {1}
for line in all_lines:
if line.strip(): # ignore any empty lines
index_of_first_comma = line.find(',')
command_name = line[:index_of_first_comma]
json_string = line[index_of_first_comma+1:]
telemetry_dict = json.loads(json_string)
return_tuples.append((command_name, telemetry_dict))
return return_tuples
:return: None
"""
telemetry_tuples = read_telemetry_entries()
telemetry_client = TelemetryClient(INSTRUMENTATION)
for tup in telemetry_tuples:
telemetry_client.track_event(tup[0], tup[1])
# This will never end if there is no internet connection, for example.
telemetry_client.flush()
# After send has completed, delete the file
try:
os.remove(TELEMETRY_FILE_PATH)
except: # pylint: disable=bare-except
pass
# After send has completed, clear the telemetry retry counter
set_telemetry_send_retry_count(0)