1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
"""
Test long running process & detect memory leak
"""
import configobj
import pytest
from unittest import mock
import unittest
import resource
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
import protobix
PAYLOAD = {
"items": {
"protobix.host1": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
},
"protobix.host2": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
},
"protobix.host3": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
},
"protobix.host4": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
},
"protobix.host5": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
},
"protobix.host6": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
},
"protobix.host7": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
},
"protobix.host8": {
"my.protobix.item.int": 0,
"my.protobix.item.string": 1
}
},
"lld": {
'protobix.host1': {
'my.protobix.lld_item1': [
{ '{#PBX_LLD_KEY11}': 0,
'{#PBX_LLD_KEY12}': 'lld string' },
{ '{#PBX_LLD_KEY11}': 1,
'{#PBX_LLD_KEY12}': 'another lld string' }
],
'my.protobix.lld_item2': [
{ '{#PBX_LLD_KEY21}': 10,
'{#PBX_LLD_KEY21}': 'yet an lld string' },
{ '{#PBX_LLD_KEY21}': 2,
'{#PBX_LLD_KEY21}': 'yet another lld string' }
]
},
'protobix.host2': {
'my.protobix.lld_item1': [
{ '{#PBX_LLD_KEY11}': 0,
'{#PBX_LLD_KEY12}': 'lld string' },
{ '{#PBX_LLD_KEY11}': 1,
'{#PBX_LLD_KEY12}': 'another lld string' }
],
'my.protobix.lld_item2': [
{ '{#PBX_LLD_KEY21}': 10,
'{#PBX_LLD_KEY21}': 'yet an lld string' },
{ '{#PBX_LLD_KEY21}': 2,
'{#PBX_LLD_KEY21}': 'yet another lld string' }
]
}
}
}
def long_run(data_type, debug_level):
"""
Generic long running process simulator
Used by tests below
"""
zbx_container = protobix.DataContainer()
zbx_container.debug_level = debug_level
run=1
max_run=1000
while run <= max_run:
zbx_container.data_type = data_type
zbx_container.add(PAYLOAD[data_type])
try:
zbx_container.send()
except:
pass
if run % (max_run/10) == 0 or run <=1:
usage=resource.getrusage(resource.RUSAGE_SELF)
display_memory = usage[2]*resource.getpagesize()/1000000.0
if run == 1:
initial_memory = usage[2]
display_initial_memory = usage[2]*resource.getpagesize()/1000000.0
final_memory = usage[2]
print ('Run %i: ru_maxrss=%f mb - initial=%f mb' % (
run, (display_memory), display_initial_memory
))
run += 1
return initial_memory, final_memory
memory_leak_matrix = (
('items', 2),
('items', 4),
('lld', 2),
('lld', 4)
)
@pytest.mark.parametrize(('data_type','debug_level'), memory_leak_matrix)
def test_long_run_for_memory_leak(data_type, debug_level):
"""
Simulate long running process with and without debug
and control memory usage
"""
initial_memory, final_memory = long_run(data_type, debug_level)
assert initial_memory == final_memory
|
