@@ -45,94 +45,81 @@ metadata:
4545 name : test-log-aggregation
4646data :
4747 test.py : |
48- import requests
48+ import json
49+ import subprocess
4950
5051
51- def check_sent_events():
52- response = requests.post(
53- 'http://opensearch-vector-aggregator:8686/graphql',
54- json={
55- 'query': """
56- {
57- transforms(first:100) {
58- nodes {
59- componentId
60- metrics {
61- sentEventsTotal {
62- sentEventsTotal
63- }
64- }
65- }
66- }
67- }
68- """
69- }
52+ def grpc_get_components(host):
53+ """Call the Vector gRPC observability API and return the parsed result."""
54+ response = subprocess.run(
55+ [
56+ "grpcurl",
57+ "-plaintext",
58+ "-d",
59+ '{"limit": 100}',
60+ host,
61+ "vector.observability.v1.ObservabilityService/GetComponents",
62+ ],
63+ capture_output=True,
64+ text=True,
65+ check=True,
66+ timeout=20,
7067 )
68+ result = json.loads(response.stdout)
69+ # TODO: Remove this debug dump once the gRPC response shape is confirmed
70+ print(f"gRPC response from {host}:")
71+ print(json.dumps(result, indent=2))
72+ return result
73+
7174
72- assert response.status_code == 200, \
73- 'Cannot access the API of the vector aggregator.'
75+ def check_sent_events():
76+ result = grpc_get_components("opensearch-vector-aggregator:8686")
77+ components = result.get("components", [])
78+ transforms = [
79+ c for c in components if c.get("componentType") == "COMPONENT_TYPE_TRANSFORM"
80+ ]
7481
75- result = response.json()
82+ assert len(transforms) > 0, "No transform components found"
7683
77- transforms = result['data']['transforms']['nodes']
7884 for transform in transforms:
79- sentEvents = transform[' metrics'][' sentEventsTotal' ]
80- componentId = transform[' componentId' ]
85+ sentEvents = transform[" metrics"][" sentEventsTotal" ]
86+ componentId = transform[" componentId" ]
8187
82- if componentId == ' filteredInvalidEvents' :
83- assert sentEvents is None or \
84- sentEvents['sentEventsTotal'] == 0, \
85- 'Invalid log events were sent.'
88+ if componentId == " filteredInvalidEvents" :
89+ assert sentEvents is None or int(sentEvents) == 0, (
90+ "Invalid log events were sent."
91+ )
8692 else:
87- assert sentEvents is not None and \
88- sentEvents['sentEventsTotal'] > 0, \
89- f'No events were sent in "{componentId}".'
93+ assert sentEvents is not None and int(sentEvents) > 0, (
94+ f'No events were sent in "{componentId}".'
95+ )
9096
9197
9298 def check_log_file_rollover():
93- response = requests.post(
94- 'http://opensearch-nodes-automatic-vector:8686/graphql',
95- json={
96- 'query': """
97- {
98- sources {
99- nodes {
100- componentId
101- metrics {
102- receivedBytesTotal {
103- receivedBytesTotal
104- }
105- }
106- }
107- }
108- }
109- """
110- }
111- )
112-
113- assert response.status_code == 200, \
114- 'Cannot access the API of the vector agent.'
115-
116- result = response.json()
99+ result = grpc_get_components("opensearch-nodes-automatic-vector:8686")
100+ components = result.get("components", [])
101+ sources = [
102+ c for c in components if c.get("componentType") == "COMPONENT_TYPE_SOURCE"
103+ ]
117104
118- sources = result['data']['sources']['nodes']
119105 for source in sources:
120- receivedBytes = source[' metrics'][' receivedBytesTotal' ]
121- componentId = source[' componentId' ]
106+ receivedBytes = source[" metrics"][" receivedBytesTotal" ]
107+ componentId = source[" componentId" ]
122108
123- if componentId == ' files_opensearch_server' :
109+ if componentId == " files_opensearch_server" :
124110 assert receivedBytes is not None
125- receivedBytes = receivedBytes['receivedBytesTotal']
111+ receivedBytes = int( receivedBytes)
126112 MAX_LOG_FILE_SIZE = 5_500_000
127113 expectedBytes = 2 * MAX_LOG_FILE_SIZE
128- assert receivedBytes >= expectedBytes, \
129- 'Log file rollover did not yet happen twice ' \
130- f'({receivedBytes:,.0f} Bytes of {expectedBytes:,d} Bytes received). ' \
131- 'The first rollover requires write permission to rename the log file, ' \
132- 'the second rollover additionally requires delete permission to remove the old log file.'
114+ assert receivedBytes >= expectedBytes, (
115+ "Log file rollover did not yet happen twice "
116+ f"({receivedBytes:,.0f} Bytes of {expectedBytes:,d} Bytes received). "
117+ "The first rollover requires write permission to rename the log file, "
118+ "the second rollover additionally requires delete permission to remove the old log file."
119+ )
133120
134121
135- if __name__ == ' __main__' :
122+ if __name__ == " __main__" :
136123 check_sent_events()
137124 check_log_file_rollover()
138- print(' Test successful!' )
125+ print(" Test successful!" )
0 commit comments