src/rrd_client.[ch]: Impelemt `rrdc_flush'.
[rrdtool.git] / src / rrd_client.c
1 /**
2  * RRDTool - src/rrd_client.c
3  * Copyright (C) 2008 Florian octo Forster
4  *
5  * This program is free software; you can redistribute it and/or modify it
6  * under the terms of the GNU General Public License as published by the
7  * Free Software Foundation; only version 2 of the License is applicable.
8  *
9  * This program is distributed in the hope that it will be useful, but
10  * WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
12  * General Public License for more details.
13  *
14  * You should have received a copy of the GNU General Public License along
15  * with this program; if not, write to the Free Software Foundation, Inc.,
16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
17  *
18  * Authors:
19  *   Florian octo Forster <octo at verplant.org>
20  **/
21
22 #include "rrd.h"
23 #include "rrd_client.h"
24
25 #include <stdlib.h>
26 #include <string.h>
27 #include <errno.h>
28 #include <assert.h>
29 #include <pthread.h>
30 #include <sys/types.h>
31 #include <sys/socket.h>
32 #include <sys/un.h>
33 #include <netdb.h>
34
35 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
36 static int sd = -1;
37 #if 0
38 static FILE *sh;
39 #endif
40
41 static ssize_t sread (void *buffer_void, size_t buffer_size) /* {{{ */
42 {
43   char    *buffer;
44   size_t   buffer_used;
45   size_t   buffer_free;
46   ssize_t  status;
47
48   buffer       = (char *) buffer_void;
49   buffer_used  = 0;
50   buffer_free  = buffer_size;
51
52   while (buffer_free > 0)
53   {
54     status = read (sd, buffer + buffer_used, buffer_free);
55     if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR)))
56       continue;
57
58     if (status < 0)
59       return (-1);
60
61     if (status == 0)
62     {
63       close (sd);
64       sd = -1;
65 #if 0
66       sh = NULL;
67 #endif
68       errno = EPROTO;
69       return (-1);
70     }
71
72     assert ((0 > status) || (buffer_free >= (size_t) status));
73
74     buffer_free = buffer_free - status;
75     buffer_used = buffer_used + status;
76
77     if (buffer[buffer_used - 1] == '\n')
78       break;
79   }
80
81   if (buffer[buffer_used - 1] != '\n')
82   {
83     errno = ENOBUFS;
84     return (-1);
85   }
86
87   buffer[buffer_used - 1] = 0;
88   return (buffer_used);
89 } /* }}} ssize_t sread */
90
91 static ssize_t swrite (const void *buf, size_t count) /* {{{ */
92 {
93   const char *ptr;
94   size_t      nleft;
95   ssize_t     status;
96
97   ptr   = (const char *) buf;
98   nleft = count;
99
100   while (nleft > 0)
101   {
102     status = write (sd, (const void *) ptr, nleft);
103
104     if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR)))
105       continue;
106
107     if (status < 0)
108       return (status);
109
110     nleft = nleft - status;
111     ptr   = ptr   + status;
112   }
113
114   return (0);
115 } /* }}} ssize_t swrite */
116
117 static int buffer_add_string (const char *str, /* {{{ */
118     char **buffer_ret, size_t *buffer_size_ret)
119 {
120   char *buffer;
121   size_t buffer_size;
122   size_t buffer_pos;
123   size_t i;
124   int status;
125
126   buffer = *buffer_ret;
127   buffer_size = *buffer_size_ret;
128   buffer_pos = 0;
129
130   i = 0;
131   status = -1;
132   while (buffer_pos < buffer_size)
133   {
134     if (str[i] == 0)
135     {
136       buffer[buffer_pos] = ' ';
137       buffer_pos++;
138       status = 0;
139       break;
140     }
141     else if ((str[i] == ' ') || (str[i] == '\\'))
142     {
143       if (buffer_pos >= (buffer_size - 1))
144         break;
145       buffer[buffer_pos] = '\\';
146       buffer_pos++;
147       buffer[buffer_pos] = str[i];
148       buffer_pos++;
149     }
150     else
151     {
152       buffer[buffer_pos] = str[i];
153       buffer_pos++;
154     }
155     i++;
156   } /* while (buffer_pos < buffer_size) */
157
158   if (status != 0)
159     return (-1);
160
161   *buffer_ret = buffer + buffer_pos;
162   *buffer_size_ret = buffer_size - buffer_pos;
163
164   return (0);
165 } /* }}} int buffer_add_string */
166
167 static int buffer_add_value (const char *value, /* {{{ */
168     char **buffer_ret, size_t *buffer_size_ret)
169 {
170   char temp[4096];
171
172   if (strncmp (value, "N:", 2) == 0)
173     snprintf (temp, sizeof (temp), "%lu:%s",
174         (unsigned long) time (NULL), value + 2);
175   else
176     strncpy (temp, value, sizeof (temp));
177   temp[sizeof (temp) - 1] = 0;
178
179   return (buffer_add_string (temp, buffer_ret, buffer_size_ret));
180 } /* }}} int buffer_add_value */
181
182 static int rrdc_connect_unix (const char *path) /* {{{ */
183 {
184   struct sockaddr_un sa;
185   int status;
186
187   if (path == NULL)
188     path = RRDD_SOCK_PATH;
189
190   pthread_mutex_lock (&lock);
191
192   if (sd >= 0)
193   {
194     pthread_mutex_unlock (&lock);
195     return (0);
196   }
197
198   sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
199   if (sd < 0)
200   {
201     status = errno;
202     pthread_mutex_unlock (&lock);
203     return (status);
204   }
205
206   memset (&sa, 0, sizeof (sa));
207   sa.sun_family = AF_UNIX;
208   strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
209
210   status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
211   if (status != 0)
212   {
213     status = errno;
214     pthread_mutex_unlock (&lock);
215     return (status);
216   }
217
218 #if 0
219   sh = fdopen (sd, "w+");
220   if (sh == NULL)
221   {
222     status = errno;
223     close (sd);
224     sd = -1;
225     pthread_mutex_unlock (&lock);
226     return (status);
227   }
228 #endif
229
230   pthread_mutex_unlock (&lock);
231
232   return (0);
233 } /* }}} int rrdc_connect_unix */
234
235 int rrdc_connect (const char *addr) /* {{{ */
236 {
237   struct addrinfo ai_hints;
238   struct addrinfo *ai_res;
239   struct addrinfo *ai_ptr;
240   int status;
241
242   if (addr == NULL)
243     addr = RRDD_SOCK_PATH;
244
245   if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
246     return (rrdc_connect_unix (addr + strlen ("unix:")));
247   else if (addr[0] == '/')
248     return (rrdc_connect_unix (addr));
249
250   pthread_mutex_lock (&lock);
251
252   if (sd >= 0)
253   {
254     pthread_mutex_unlock (&lock);
255     return (0);
256   }
257
258   memset (&ai_hints, 0, sizeof (ai_hints));
259   ai_hints.ai_flags = 0;
260 #ifdef AI_ADDRCONFIG
261   ai_hints.ai_flags |= AI_ADDRCONFIG;
262 #endif
263   ai_hints.ai_family = AF_UNSPEC;
264   ai_hints.ai_socktype = SOCK_STREAM;
265
266   ai_res = NULL;
267   status = getaddrinfo (addr, DEFAULT_PORT, &ai_hints, &ai_res);
268   if (status != 0)
269   {
270     pthread_mutex_unlock (&lock);
271     return (status);
272   }
273
274   for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
275   {
276     sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
277     if (sd < 0)
278     {
279       status = errno;
280       sd = -1;
281       continue;
282     }
283
284     status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
285     if (status != 0)
286     {
287       status = errno;
288       close (sd);
289       sd = -1;
290       continue;
291     }
292
293 #if 0
294     sh = fdopen (sd, "w+");
295     if (sh == NULL)
296     {
297       status = errno;
298       close (sd);
299       sd = -1;
300       continue;
301     } 
302 #endif
303
304     assert (status == 0);
305     break;
306   } /* for (ai_ptr) */
307   pthread_mutex_unlock (&lock);
308
309   return (status);
310 } /* }}} int rrdc_connect */
311
312 int rrdc_disconnect (void) /* {{{ */
313 {
314   int status;
315
316   pthread_mutex_lock (&lock);
317
318   if (sd < 0)
319   {
320     pthread_mutex_unlock (&lock);
321     return (-1);
322   }
323
324 #if 0
325   status = fclose (sh);
326   if (status != 0)
327     status = errno;
328
329   sh = NULL;
330 #else
331   status = 0;
332 #endif
333   sd = -1;
334
335   pthread_mutex_unlock (&lock);
336
337   return (status);
338 } /* }}} int rrdc_disconnect */
339
340 int rrdc_update (const char *filename, int values_num, /* {{{ */
341                 const char * const *values)
342 {
343   char buffer[4096];
344   char *buffer_ptr;
345   size_t buffer_free;
346   size_t buffer_size;
347   int status;
348   int i;
349
350   memset (buffer, 0, sizeof (buffer));
351   buffer_ptr = &buffer[0];
352   buffer_free = sizeof (buffer);
353
354   status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
355   if (status != 0)
356     return (ENOBUFS);
357
358   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
359   if (status != 0)
360     return (ENOBUFS);
361
362   for (i = 0; i < values_num; i++)
363   {
364     status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
365     if (status != 0)
366       return (ENOBUFS);
367   }
368
369   assert (buffer_free < sizeof (buffer));
370   buffer_size = sizeof (buffer) - buffer_free;
371   assert (buffer[buffer_size - 1] == ' ');
372   buffer[buffer_size - 1] = '\n';
373
374   pthread_mutex_lock (&lock);
375
376   if (sd < 0)
377   {
378     pthread_mutex_unlock (&lock);
379     return (ENOTCONN);
380   }
381
382   status = swrite (buffer, buffer_size);
383   if (status != 0)
384   {
385     pthread_mutex_unlock (&lock);
386     return (status);
387   }
388
389   status = sread (buffer, sizeof (buffer));
390   if (status < 0)
391   {
392     status = errno;
393     pthread_mutex_unlock (&lock);
394     return (status);
395   }
396   else if (status == 0)
397   {
398     pthread_mutex_unlock (&lock);
399     return (ENODATA);
400   }
401
402   pthread_mutex_unlock (&lock);
403
404   status = atoi (buffer);
405   return (status);
406 } /* }}} int rrd_update_daemon */
407
408 int rrdc_flush (const char *filename) /* {{{ */
409 {
410   char buffer[4096];
411   char *buffer_ptr;
412   size_t buffer_free;
413   size_t buffer_size;
414   int status;
415
416   if (filename == NULL)
417     return (-1);
418
419   memset (buffer, 0, sizeof (buffer));
420   buffer_ptr = &buffer[0];
421   buffer_free = sizeof (buffer);
422
423   status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
424   if (status != 0)
425     return (ENOBUFS);
426
427   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
428   if (status != 0)
429     return (ENOBUFS);
430
431   assert (buffer_free < sizeof (buffer));
432   buffer_size = sizeof (buffer) - buffer_free;
433   assert (buffer[buffer_size - 1] == ' ');
434   buffer[buffer_size - 1] = '\n';
435
436   pthread_mutex_lock (&lock);
437
438   if (sd < 0)
439   {
440     pthread_mutex_unlock (&lock);
441     return (ENOTCONN);
442   }
443
444   status = swrite (buffer, buffer_size);
445   if (status != 0)
446   {
447     pthread_mutex_unlock (&lock);
448     return (status);
449   }
450
451   status = sread (buffer, sizeof (buffer));
452   if (status < 0)
453   {
454     status = errno;
455     pthread_mutex_unlock (&lock);
456     return (status);
457   }
458   else if (status == 0)
459   {
460     pthread_mutex_unlock (&lock);
461     return (ENODATA);
462   }
463
464   pthread_mutex_unlock (&lock);
465
466   status = atoi (buffer);
467   return (status);
468 } /* }}} int rrdc_flush */
469
470 /*
471  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
472  */