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 <android-base/threads.h>
19 #include <cutils/fs.h>
20 #include <cutils/sockets.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 static pthread_mutex_t log_init_lock = PTHREAD_MUTEX_INITIALIZER;
40 static atomic_int dropped = 0;
41 static atomic_int log_error = 0;
42 static atomic_int atom_tag = 0;
43
statsd_writer_init_lock()44 void statsd_writer_init_lock() {
45 /*
46 * If we trigger a signal handler in the middle of locked activity and the
47 * signal handler logs a message, we could get into a deadlock state.
48 */
49 pthread_mutex_lock(&log_init_lock);
50 }
51
statd_writer_trylock()52 int statd_writer_trylock() {
53 return pthread_mutex_trylock(&log_init_lock);
54 }
55
statsd_writer_init_unlock()56 void statsd_writer_init_unlock() {
57 pthread_mutex_unlock(&log_init_lock);
58 }
59
60 static int statsdAvailable();
61 static int statsdOpen();
62 static void statsdClose();
63 static int statsdWrite(struct timespec* ts, struct iovec* vec, size_t nr);
64 static void statsdNoteDrop(int error, int tag);
65
66 struct android_log_transport_write statsdLoggerWrite = {
67 .name = "statsd",
68 .sock = -EBADF,
69 .available = statsdAvailable,
70 .open = statsdOpen,
71 .close = statsdClose,
72 .write = statsdWrite,
73 .noteDrop = statsdNoteDrop,
74 };
75
76 /* log_init_lock assumed */
statsdOpen()77 static int statsdOpen() {
78 int i, ret = 0;
79
80 i = atomic_load(&statsdLoggerWrite.sock);
81 if (i < 0) {
82 int flags = SOCK_DGRAM;
83 #ifdef SOCK_CLOEXEC
84 flags |= SOCK_CLOEXEC;
85 #endif
86 #ifdef SOCK_NONBLOCK
87 flags |= SOCK_NONBLOCK;
88 #endif
89 int sock = TEMP_FAILURE_RETRY(socket(PF_UNIX, flags, 0));
90 if (sock < 0) {
91 ret = -errno;
92 } else {
93 int sndbuf = 1 * 1024 * 1024; // set max send buffer size 1MB
94 socklen_t bufLen = sizeof(sndbuf);
95 // SO_RCVBUF does not have an effect on unix domain socket, but SO_SNDBUF does.
96 // Proceed to connect even setsockopt fails.
97 setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &sndbuf, bufLen);
98 struct sockaddr_un un;
99 memset(&un, 0, sizeof(struct sockaddr_un));
100 un.sun_family = AF_UNIX;
101 strcpy(un.sun_path, "/dev/socket/statsdw");
102
103 if (TEMP_FAILURE_RETRY(
104 connect(sock, (struct sockaddr*)&un, sizeof(struct sockaddr_un))) < 0) {
105 ret = -errno;
106 switch (ret) {
107 case -ENOTCONN:
108 case -ECONNREFUSED:
109 case -ENOENT:
110 i = atomic_exchange(&statsdLoggerWrite.sock, ret);
111 break;
112 default:
113 break;
114 }
115 close(sock);
116 } else {
117 ret = atomic_exchange(&statsdLoggerWrite.sock, sock);
118 if ((ret >= 0) && (ret != sock)) {
119 close(ret);
120 }
121 ret = 0;
122 }
123 }
124 }
125
126 return ret;
127 }
128
__statsdClose(int negative_errno)129 static void __statsdClose(int negative_errno) {
130 int sock = atomic_exchange(&statsdLoggerWrite.sock, negative_errno);
131 if (sock >= 0) {
132 close(sock);
133 }
134 }
135
statsdClose()136 static void statsdClose() {
137 __statsdClose(-EBADF);
138 }
139
statsdAvailable()140 static int statsdAvailable() {
141 if (atomic_load(&statsdLoggerWrite.sock) < 0) {
142 if (access("/dev/socket/statsdw", W_OK) == 0) {
143 return 0;
144 }
145 return -EBADF;
146 }
147 return 1;
148 }
149
statsdNoteDrop(int error,int tag)150 static void statsdNoteDrop(int error, int tag) {
151 atomic_fetch_add_explicit(&dropped, 1, memory_order_relaxed);
152 atomic_exchange_explicit(&log_error, error, memory_order_relaxed);
153 atomic_exchange_explicit(&atom_tag, tag, memory_order_relaxed);
154 }
155
statsdWrite(struct timespec * ts,struct iovec * vec,size_t nr)156 static int statsdWrite(struct timespec* ts, struct iovec* vec, size_t nr) {
157 ssize_t ret;
158 int sock;
159 static const unsigned headerLength = 1;
160 struct iovec newVec[nr + headerLength];
161 android_log_header_t header;
162 size_t i, payloadSize;
163
164 sock = atomic_load(&statsdLoggerWrite.sock);
165 if (sock < 0) switch (sock) {
166 case -ENOTCONN:
167 case -ECONNREFUSED:
168 case -ENOENT:
169 break;
170 default:
171 return -EBADF;
172 }
173 /*
174 * struct {
175 * // what we provide to socket
176 * android_log_header_t header;
177 * // caller provides
178 * union {
179 * struct {
180 * char prio;
181 * char payload[];
182 * } string;
183 * struct {
184 * uint32_t tag
185 * char payload[];
186 * } binary;
187 * };
188 * };
189 */
190
191 header.tid = android::base::GetThreadId();
192 header.realtime.tv_sec = ts->tv_sec;
193 header.realtime.tv_nsec = ts->tv_nsec;
194
195 newVec[0].iov_base = (unsigned char*)&header;
196 newVec[0].iov_len = sizeof(header);
197
198 // If we dropped events before, try to tell statsd.
199 if (sock >= 0) {
200 int32_t snapshot = atomic_exchange_explicit(&dropped, 0, memory_order_relaxed);
201 if (snapshot) {
202 android_log_event_long_t buffer;
203 header.id = LOG_ID_STATS;
204 // store the last log error in the tag field. This tag field is not used by statsd.
205 buffer.header.tag = atomic_load(&log_error);
206 buffer.payload.type = EVENT_TYPE_LONG;
207 // format:
208 // |atom_tag|dropped_count|
209 int64_t composed_long = atomic_load(&atom_tag);
210 // Send 2 int32's via an int64.
211 composed_long = ((composed_long << 32) | ((int64_t)snapshot));
212 buffer.payload.data = composed_long;
213
214 newVec[headerLength].iov_base = &buffer;
215 newVec[headerLength].iov_len = sizeof(buffer);
216
217 ret = TEMP_FAILURE_RETRY(writev(sock, newVec, 2));
218 if (ret != (ssize_t)(sizeof(header) + sizeof(buffer))) {
219 atomic_fetch_add_explicit(&dropped, snapshot, memory_order_relaxed);
220 }
221 }
222 }
223
224 header.id = LOG_ID_STATS;
225
226 for (payloadSize = 0, i = headerLength; i < nr + headerLength; i++) {
227 newVec[i].iov_base = vec[i - headerLength].iov_base;
228 payloadSize += newVec[i].iov_len = vec[i - headerLength].iov_len;
229
230 if (payloadSize > LOGGER_ENTRY_MAX_PAYLOAD) {
231 newVec[i].iov_len -= payloadSize - LOGGER_ENTRY_MAX_PAYLOAD;
232 if (newVec[i].iov_len) {
233 ++i;
234 }
235 break;
236 }
237 }
238
239 /*
240 * The write below could be lost, but will never block.
241 *
242 * ENOTCONN occurs if statsd has died.
243 * ENOENT occurs if statsd is not running and socket is missing.
244 * ECONNREFUSED occurs if we can not reconnect to statsd.
245 * EAGAIN occurs if statsd is overloaded.
246 */
247 if (sock < 0) {
248 ret = sock;
249 } else {
250 ret = TEMP_FAILURE_RETRY(writev(sock, newVec, i));
251 if (ret < 0) {
252 ret = -errno;
253 }
254 }
255 switch (ret) {
256 case -ENOTCONN:
257 case -ECONNREFUSED:
258 case -ENOENT:
259 if (statd_writer_trylock()) {
260 return ret; /* in a signal handler? try again when less stressed
261 */
262 }
263 __statsdClose(ret);
264 ret = statsdOpen();
265 statsd_writer_init_unlock();
266
267 if (ret < 0) {
268 return ret;
269 }
270
271 ret = TEMP_FAILURE_RETRY(writev(atomic_load(&statsdLoggerWrite.sock), newVec, i));
272 if (ret < 0) {
273 ret = -errno;
274 }
275 break;
276 default:
277 break;
278 }
279
280 if (ret > (ssize_t)sizeof(header)) {
281 ret -= sizeof(header);
282 }
283
284 return ret;
285 }
286