Line data Source code
1 : /*
2 : * If not stated otherwise in this file or this component's LICENSE file the
3 : * following copyright and licenses apply:
4 : *
5 : * Copyright 2026 Sky UK
6 : *
7 : * Licensed under the Apache License, Version 2.0 (the "License");
8 : * you may not use this file except in compliance with the License.
9 : * You may obtain 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,
15 : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 : * See the License for the specific language governing permissions and
17 : * limitations under the License.
18 : */
19 :
20 : #include "PrivateMetricsService.h"
21 : #include "RialtoServerLogging.h"
22 : #include <cinttypes>
23 : #include <cstdio>
24 : #include <fstream>
25 : #include <string>
26 : #include <utility>
27 :
28 : namespace firebolt::rialto::server::service
29 : {
30 12 : PrivateMetricsService::PrivateMetricsService(
31 12 : std::shared_ptr<firebolt::rialto::server::IMetricsCollectorFactory> collectorFactory)
32 12 : : m_collectorFactory{std::move(collectorFactory)}
33 : {
34 : }
35 :
36 19 : PrivateMetricsService::~PrivateMetricsService()
37 : {
38 12 : std::lock_guard<std::mutex> lock{m_mutex};
39 12 : m_collectors.clear();
40 19 : }
41 :
42 5 : void PrivateMetricsService::clientReady(int clientId,
43 : const std::shared_ptr<firebolt::rialto::server::IMetricsCollectorClient> &client)
44 : {
45 5 : std::lock_guard<std::mutex> lock{m_mutex};
46 5 : if (!m_collectorFactory)
47 : {
48 0 : RIALTO_SERVER_LOG_ERROR("MetricsCollectorFactory is null; cannot create MetricsCollector for client %d",
49 : clientId);
50 0 : return;
51 : }
52 5 : auto collector = m_collectorFactory->create(clientId, client, m_currentApplicationState);
53 5 : if (collector)
54 : {
55 4 : m_collectors.emplace(clientId, std::move(collector));
56 4 : RIALTO_SERVER_LOG_INFO("MetricsCollector created for client %d", clientId);
57 : }
58 : else
59 : {
60 1 : RIALTO_SERVER_LOG_ERROR("Failed to create MetricsCollector for client %d", clientId);
61 : }
62 5 : }
63 :
64 2 : void PrivateMetricsService::clientDisconnected(int clientId)
65 : {
66 2 : std::lock_guard<std::mutex> lock{m_mutex};
67 2 : auto iter = m_collectors.find(clientId);
68 2 : if (iter != m_collectors.end())
69 : {
70 1 : m_collectors.erase(iter);
71 1 : RIALTO_SERVER_LOG_INFO("MetricsCollector destroyed for client %d", clientId);
72 : }
73 2 : }
74 :
75 2 : void PrivateMetricsService::reportMetrics(int clientId, const firebolt::rialto::server::ClientMetricsData &metrics)
76 : {
77 2 : std::lock_guard<std::mutex> lock{m_mutex};
78 2 : auto iter = m_collectors.find(clientId);
79 2 : if (iter != m_collectors.end())
80 : {
81 1 : iter->second->processMetrics(metrics);
82 : }
83 : else
84 : {
85 1 : RIALTO_SERVER_LOG_WARN("reportMetrics for unknown client %d", clientId);
86 : }
87 2 : }
88 :
89 2 : void PrivateMetricsService::notifyPlaybackStateChanged(int sessionId, PlaybackState oldState, PlaybackState newState)
90 : {
91 2 : std::lock_guard<std::mutex> lock{m_mutex};
92 3 : for (auto &[clientId, collector] : m_collectors)
93 : {
94 1 : collector->notifyPlaybackStateChanged(sessionId, oldState, newState);
95 : }
96 2 : }
97 :
98 1 : void PrivateMetricsService::notifyWebAudioPlayerStateChanged(int handle, WebAudioPlayerState oldState,
99 : WebAudioPlayerState newState)
100 : {
101 1 : std::lock_guard<std::mutex> lock{m_mutex};
102 2 : for (auto &[clientId, collector] : m_collectors)
103 : {
104 1 : collector->notifyWebAudioPlayerStateChanged(handle, oldState, newState);
105 : }
106 : }
107 :
108 5 : void PrivateMetricsService::notifyApplicationStateChanged(ApplicationState newState)
109 : {
110 : {
111 5 : std::lock_guard<std::mutex> lock{m_mutex};
112 : // Keep transition history here so upstream services only report the new application state.
113 5 : const auto oldState{m_currentApplicationState};
114 5 : m_currentApplicationState = newState;
115 8 : for (auto &[clientId, collector] : m_collectors)
116 : {
117 3 : collector->notifyApplicationStateChanged(oldState, newState);
118 : }
119 5 : }
120 :
121 : // When transitioning to INACTIVE, record a server-side memory snapshot.
122 : // At this point, pipelines and shared memory have already been freed but
123 : // no client may be connected to supply a full sample — so we read the
124 : // server's own memory directly. The collector mutex is released first so
125 : // filesystem access cannot block metrics processing.
126 5 : if (newState == ApplicationState::INACTIVE)
127 : {
128 2 : std::uint64_t serverMemoryKb{0};
129 : {
130 2 : std::ifstream status{"/proc/self/status"};
131 2 : std::string line;
132 46 : while (std::getline(status, line))
133 : {
134 46 : if (line.rfind("VmRSS:", 0) == 0)
135 : {
136 2 : std::sscanf(line.c_str(), "VmRSS: %" SCNu64, &serverMemoryKb);
137 2 : break;
138 : }
139 : }
140 : }
141 :
142 2 : std::uint64_t cgroupMemoryUsageKb{0};
143 : {
144 2 : auto readFileValue = [](const std::string &path) -> std::uint64_t
145 : {
146 2 : std::ifstream file{path};
147 2 : if (!file.is_open())
148 : {
149 0 : return 0;
150 : }
151 2 : std::string content;
152 2 : if (!std::getline(file, content) || content.empty() || content == "max")
153 : {
154 0 : return 0;
155 : }
156 2 : std::uint64_t value{0};
157 2 : if (std::sscanf(content.c_str(), "%" SCNu64, &value) == 1)
158 : {
159 2 : return value;
160 : }
161 0 : return 0;
162 2 : };
163 :
164 : // Resolve the process's cgroup path from /proc/self/cgroup
165 2 : std::ifstream cgroupFile{"/proc/self/cgroup"};
166 2 : std::string cgroupBase;
167 2 : if (cgroupFile.is_open())
168 : {
169 2 : std::string line;
170 2 : while (std::getline(cgroupFile, line))
171 : {
172 2 : if (line.rfind("0::", 0) == 0)
173 : {
174 2 : std::string relativePath{line.substr(3)};
175 2 : if (!relativePath.empty() && relativePath != "/")
176 : {
177 2 : cgroupBase = "/sys/fs/cgroup" + relativePath;
178 : }
179 : else
180 : {
181 0 : cgroupBase = "/sys/fs/cgroup";
182 : }
183 2 : break;
184 : }
185 : }
186 : }
187 :
188 2 : std::uint64_t usageBytes{0};
189 2 : if (!cgroupBase.empty())
190 : {
191 2 : usageBytes = readFileValue(cgroupBase + "/memory.current");
192 : }
193 2 : if (usageBytes == 0)
194 : {
195 0 : usageBytes = readFileValue("/sys/fs/cgroup/memory/memory.usage_in_bytes");
196 : }
197 2 : cgroupMemoryUsageKb = usageBytes / 1024;
198 : }
199 :
200 : // Read smaps_rollup to split private-dirty heap from file-backed libs.
201 2 : std::uint64_t anonKb{0}, sharedCleanKb{0}, privateCleanKb{0}, privateDirtyKb{0};
202 : {
203 2 : std::ifstream smaps{"/proc/self/smaps_rollup"};
204 2 : std::string sline;
205 48 : while (std::getline(smaps, sline))
206 : {
207 46 : if (sline.rfind("Anonymous:", 0) == 0)
208 2 : std::sscanf(sline.c_str(), "Anonymous: %" SCNu64, &anonKb);
209 44 : else if (sline.rfind("Shared_Clean:", 0) == 0)
210 2 : std::sscanf(sline.c_str(), "Shared_Clean: %" SCNu64, &sharedCleanKb);
211 42 : else if (sline.rfind("Private_Clean:", 0) == 0)
212 2 : std::sscanf(sline.c_str(), "Private_Clean: %" SCNu64, &privateCleanKb);
213 40 : else if (sline.rfind("Private_Dirty:", 0) == 0)
214 2 : std::sscanf(sline.c_str(), "Private_Dirty: %" SCNu64, &privateDirtyKb);
215 : }
216 : }
217 2 : RIALTO_SERVER_LOG_MIL("Metrics: INACTIVE memory snapshot — server_mem_kb=%" PRIu64 ", cgroup_mem_kb=%" PRIu64
218 : ", anon_kb=%" PRIu64 ", private_dirty_kb=%" PRIu64 ", private_clean_kb=%" PRIu64
219 : ", shared_clean_kb=%" PRIu64,
220 : serverMemoryKb, cgroupMemoryUsageKb, anonKb, privateDirtyKb, privateCleanKb, sharedCleanKb);
221 : }
222 5 : }
223 : } // namespace firebolt::rialto::server::service
|
