1 /*
2 * Copyright (C) 2018, The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16 #include "statsd_writer.h"
17
18 #include <cutils/fs.h>
19 #include <cutils/sockets.h>
20 #include <cutils/threads.h>
21 #include <errno.h>
22 #include <fcntl.h>
23 #include <inttypes.h>
24 #include <poll.h>
25 #include <private/android_filesystem_config.h>
26 #include <private/android_logger.h>
27 #include <stdarg.h>
28 #include <stdatomic.h>
29 #include <stdio.h>
30 #include <stdlib.h>
31 #include <string.h>
32 #include <sys/stat.h>
33 #include <sys/types.h>
34 #include <sys/uio.h>
35 #include <sys/un.h>
36 #include <time.h>
37 #include <unistd.h>
38
39 /* branchless on many architectures. */
40 #define min(x, y) ((y) ^ (((x) ^ (y)) & -((x) < (y))))
41
42 #ifndef htole32
43 #if __BYTE_ORDER == __LITTLE_ENDIAN
44 #define htole32(x) (x)
45 #else
46 #define htole32(x) __bswap_32(x)
47 #endif
48 #endif
49
50 #ifndef htole64
51 #if __BYTE_ORDER == __LITTLE_ENDIAN
52 #define htole64(x) (x)
53 #else
54 #define htole64(x) __bswap_64(x)
55 #endif
56 #endif
57
58 static pthread_mutex_t log_init_lock = PTHREAD_MUTEX_INITIALIZER;
59 static atomic_int dropped = 0;
60 static atomic_int log_error = 0;
61 static atomic_int atom_tag = 0;
62
statsd_writer_init_lock()63 void statsd_writer_init_lock() {
64 /*
65 * If we trigger a signal handler in the middle of locked activity and the
66 * signal handler logs a message, we could get into a deadlock state.
67 */
68 pthread_mutex_lock(&log_init_lock);
69 }
70
statd_writer_trylock()71 int statd_writer_trylock() {
72 return pthread_mutex_trylock(&log_init_lock);
73 }
74
statsd_writer_init_unlock()75 void statsd_writer_init_unlock() {
76 pthread_mutex_unlock(&log_init_lock);
77 }
78
79 static int statsdAvailable();
80 static int statsdOpen();
81 static void statsdClose();
82 static int statsdWrite(struct timespec* ts, struct iovec* vec, size_t nr);
83 static void statsdNoteDrop();
84
85 struct android_log_transport_write statsdLoggerWrite = {
86 .name = "statsd",
87 .sock = -EBADF,
88 .available = statsdAvailable,
89 .open = statsdOpen,
90 .close = statsdClose,
91 .write = statsdWrite,
92 .noteDrop = statsdNoteDrop,
93 };
94
95 /* log_init_lock assumed */
statsdOpen()96 static int statsdOpen() {
97 int i, ret = 0;
98
99 i = atomic_load(&statsdLoggerWrite.sock);
100 if (i < 0) {
101 int flags = SOCK_DGRAM;
102 #ifdef SOCK_CLOEXEC
103 flags |= SOCK_CLOEXEC;
104 #endif
105 #ifdef SOCK_NONBLOCK
106 flags |= SOCK_NONBLOCK;
107 #endif
108 int sock = TEMP_FAILURE_RETRY(socket(PF_UNIX, flags, 0));
109 if (sock < 0) {
110 ret = -errno;
111 } else {
112 int sndbuf = 1 * 1024 * 1024; // set max send buffer size 1MB
113 socklen_t bufLen = sizeof(sndbuf);
114 // SO_RCVBUF does not have an effect on unix domain socket, but SO_SNDBUF does.
115 // Proceed to connect even setsockopt fails.
116 setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &sndbuf, bufLen);
117 struct sockaddr_un un;
118 memset(&un, 0, sizeof(struct sockaddr_un));
119 un.sun_family = AF_UNIX;
120 strcpy(un.sun_path, "/dev/socket/statsdw");
121
122 if (TEMP_FAILURE_RETRY(
123 connect(sock, (struct sockaddr*)&un, sizeof(struct sockaddr_un))) < 0) {
124 ret = -errno;
125 switch (ret) {
126 case -ENOTCONN:
127 case -ECONNREFUSED:
128 case -ENOENT:
129 i = atomic_exchange(&statsdLoggerWrite.sock, ret);
130 /* FALLTHRU */
131 default:
132 break;
133 }
134 close(sock);
135 } else {
136 ret = atomic_exchange(&statsdLoggerWrite.sock, sock);
137 if ((ret >= 0) && (ret != sock)) {
138 close(ret);
139 }
140 ret = 0;
141 }
142 }
143 }
144
145 return ret;
146 }
147
__statsdClose(int negative_errno)148 static void __statsdClose(int negative_errno) {
149 int sock = atomic_exchange(&statsdLoggerWrite.sock, negative_errno);
150 if (sock >= 0) {
151 close(sock);
152 }
153 }
154
statsdClose()155 static void statsdClose() {
156 __statsdClose(-EBADF);
157 }
158
statsdAvailable()159 static int statsdAvailable() {
160 if (atomic_load(&statsdLoggerWrite.sock) < 0) {
161 if (access("/dev/socket/statsdw", W_OK) == 0) {
162 return 0;
163 }
164 return -EBADF;
165 }
166 return 1;
167 }
168
statsdNoteDrop(int error,int tag)169 static void statsdNoteDrop(int error, int tag) {
170 atomic_fetch_add_explicit(&dropped, 1, memory_order_relaxed);
171 atomic_exchange_explicit(&log_error, error, memory_order_relaxed);
172 atomic_exchange_explicit(&atom_tag, tag, memory_order_relaxed);
173 }
174
statsdWrite(struct timespec * ts,struct iovec * vec,size_t nr)175 static int statsdWrite(struct timespec* ts, struct iovec* vec, size_t nr) {
176 ssize_t ret;
177 int sock;
178 static const unsigned headerLength = 1;
179 struct iovec newVec[nr + headerLength];
180 android_log_header_t header;
181 size_t i, payloadSize;
182
183 sock = atomic_load(&statsdLoggerWrite.sock);
184 if (sock < 0) switch (sock) {
185 case -ENOTCONN:
186 case -ECONNREFUSED:
187 case -ENOENT:
188 break;
189 default:
190 return -EBADF;
191 }
192 /*
193 * struct {
194 * // what we provide to socket
195 * android_log_header_t header;
196 * // caller provides
197 * union {
198 * struct {
199 * char prio;
200 * char payload[];
201 * } string;
202 * struct {
203 * uint32_t tag
204 * char payload[];
205 * } binary;
206 * };
207 * };
208 */
209
210 header.tid = gettid();
211 header.realtime.tv_sec = ts->tv_sec;
212 header.realtime.tv_nsec = ts->tv_nsec;
213
214 newVec[0].iov_base = (unsigned char*)&header;
215 newVec[0].iov_len = sizeof(header);
216
217 // If we dropped events before, try to tell statsd.
218 if (sock >= 0) {
219 int32_t snapshot = atomic_exchange_explicit(&dropped, 0, memory_order_relaxed);
220 if (snapshot) {
221 android_log_event_long_t buffer;
222 header.id = LOG_ID_STATS;
223 // store the last log error in the tag field. This tag field is not used by statsd.
224 buffer.header.tag = htole32(atomic_load(&log_error));
225 buffer.payload.type = EVENT_TYPE_LONG;
226 // format:
227 // |atom_tag|dropped_count|
228 int64_t composed_long = atomic_load(&atom_tag);
229 // Send 2 int32's via an int64.
230 composed_long = ((composed_long << 32) | ((int64_t)snapshot));
231 buffer.payload.data = htole64(composed_long);
232
233 newVec[headerLength].iov_base = &buffer;
234 newVec[headerLength].iov_len = sizeof(buffer);
235
236 ret = TEMP_FAILURE_RETRY(writev(sock, newVec, 2));
237 if (ret != (ssize_t)(sizeof(header) + sizeof(buffer))) {
238 atomic_fetch_add_explicit(&dropped, snapshot, memory_order_relaxed);
239 }
240 }
241 }
242
243 header.id = LOG_ID_STATS;
244
245 for (payloadSize = 0, i = headerLength; i < nr + headerLength; i++) {
246 newVec[i].iov_base = vec[i - headerLength].iov_base;
247 payloadSize += newVec[i].iov_len = vec[i - headerLength].iov_len;
248
249 if (payloadSize > LOGGER_ENTRY_MAX_PAYLOAD) {
250 newVec[i].iov_len -= payloadSize - LOGGER_ENTRY_MAX_PAYLOAD;
251 if (newVec[i].iov_len) {
252 ++i;
253 }
254 break;
255 }
256 }
257
258 /*
259 * The write below could be lost, but will never block.
260 *
261 * ENOTCONN occurs if statsd has died.
262 * ENOENT occurs if statsd is not running and socket is missing.
263 * ECONNREFUSED occurs if we can not reconnect to statsd.
264 * EAGAIN occurs if statsd is overloaded.
265 */
266 if (sock < 0) {
267 ret = sock;
268 } else {
269 ret = TEMP_FAILURE_RETRY(writev(sock, newVec, i));
270 if (ret < 0) {
271 ret = -errno;
272 }
273 }
274 switch (ret) {
275 case -ENOTCONN:
276 case -ECONNREFUSED:
277 case -ENOENT:
278 if (statd_writer_trylock()) {
279 return ret; /* in a signal handler? try again when less stressed
280 */
281 }
282 __statsdClose(ret);
283 ret = statsdOpen();
284 statsd_writer_init_unlock();
285
286 if (ret < 0) {
287 return ret;
288 }
289
290 ret = TEMP_FAILURE_RETRY(writev(atomic_load(&statsdLoggerWrite.sock), newVec, i));
291 if (ret < 0) {
292 ret = -errno;
293 }
294 /* FALLTHRU */
295 default:
296 break;
297 }
298
299 if (ret > (ssize_t)sizeof(header)) {
300 ret -= sizeof(header);
301 }
302
303 return ret;
304 }
305