?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Add replication for bucket-create requests.
[iwhd.git] / rest.c
blob36c318f02a66a6902d5b12cac6c53a2a154851d9
1 /* Copyright (C) 2010 Free Software Foundation, Inc.
2
3 This program is free software: you can redistribute it and/or modify
4 it under the terms of the GNU General Public License as published by
5 the Free Software Foundation, either version 3 of the License, or
6 (at your option) any later version.
7
8 This program is distributed in the hope that it will be useful,
9 but WITHOUT ANY WARRANTY; without even the implied warranty of
10 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11 GNU General Public License for more details.
12
13 You should have received a copy of the GNU General Public License
14 along with this program. If not, see <http://www.gnu.org/licenses/>. */
15
16 #include <config.h>
17
18 #include <error.h>
19 #include <fcntl.h>
20 #include <getopt.h>
21 #include <poll.h>
22 #include <pthread.h>
23 #include <semaphore.h>
24 #include <stdint.h>
25 #include <stdio.h>
26 #include <stdlib.h>
27 #include <string.h>
28 #include <strings.h>
29 #include <unistd.h>
30 #include <sys/stat.h>
31 #include <assert.h>
32
33 #include <microhttpd.h>
34 #include <hstor.h> /* only for ARRAY_SIZE at this point */
35 #include <curl/curl.h>
36 #include <glib.h>
37
38 #include "iwh.h"
39 #include "meta.h"
40 #include "backend.h"
41 #include "proxy.h"
42 #include "template.h"
43 #include "mpipe.h"
44 #include "state_defs.h"
45
46 /* Define-away for now. Eventually, define to gettext. */
47 #define _(msgid) (msgid)
48
49 #if defined(DEBUG)
50 #define MY_MHD_FLAGS MHD_USE_THREAD_PER_CONNECTION | MHD_USE_DEBUG
51 //#define MY_MHD_FLAGS MHD_USE_SELECT_INTERNALLY | MHD_USE_DEBUG
52 #else
53 #define MY_MHD_FLAGS MHD_USE_THREAD_PER_CONNECTION
54 #endif
55
56 extern backend_func_tbl bad_func_tbl;
57 extern backend_func_tbl s3_func_tbl;
58 extern backend_func_tbl curl_func_tbl;
59 extern backend_func_tbl fs_func_tbl;
60
61 typedef enum {
62 URL_ROOT=0, URL_BUCKET, URL_OBJECT, URL_ATTR, URL_INVAL,
63 URL_QUERY, URL_PROVLIST
64 } url_type;
65
66 typedef struct {
67 char *method;
68 url_type utype;
69 MHD_AccessHandlerCallback handler;
70 } rule;
71
72 int fs_mode = 0;
73 unsigned short my_port = MY_PORT;
74 const char *program_name;
75
76 char *(reserved_name[]) = { "_default", "_query", "_new", NULL };
77 char *(reserved_attr[]) = { "bucket", "key", "date", "etag", "loc", NULL };
78
79 backend_func_tbl *main_func_tbl = &bad_func_tbl;
80
81 void
82 free_ms (my_state *ms)
83 {
84 if (ms->cleanup & CLEANUP_CURL) {
85 curl_easy_cleanup(ms->curl);
86 }
87
88 if (ms->cleanup & CLEANUP_BUF_PTR) {
89 free(ms->pipe.data_ptr);
90 }
91
92 if (ms->cleanup & CLEANUP_POST) {
93 MHD_destroy_post_processor(ms->post);
94 }
95
96 if (ms->cleanup & CLEANUP_DICT) {
97 g_hash_table_destroy(ms->dict);
98 }
99
100 if (ms->cleanup & CLEANUP_QUERY) {
101 meta_query_stop(ms->query);
102 }
103
104 if (ms->cleanup & CLEANUP_TMPL) {
105 free(ms->gen_ctx);
106 }
107
108 if (ms->cleanup & CLEANUP_URL) {
109 free(ms->url);
110 }
111
112 free(ms);
113 }
114
115 int
116 validate_put (struct MHD_Connection *conn)
117 {
118 const char *mhdr;
119
120 if (!master_host) {
121 /* We're not a slave, so we don't care. */
122 return 1;
123 }
124
125 mhdr = MHD_lookup_connection_value(conn,MHD_HEADER_KIND,
126 "X-redhat-role");
127 return (mhdr && !strcmp(mhdr,"master"));
128 }
129
130 int
131 is_reserved (char *cand, char **resv_list)
132 {
133 int i;
134
135 for (i = 0; resv_list[i]; ++i) {
136 if (!strcmp(cand,resv_list[i])) {
137 return TRUE;
138 }
139 }
140
141 return FALSE;
142 }
143
144 int
145 validate_url (const char *url)
146 {
147 char *slash = rindex(url,'/');
148
149 if (!slash) {
150 /* There should be at least one betwixt bucket and key. */
151 return 0;
152 }
153
154 return !is_reserved(slash+1,reserved_name);
155 }
156
157 /**********
158 * The proxy has MHD on one side and CURL on the other. The CURL side is
159 * always run in a child thread. Yes, there are both context switches
160 * and copies between the threads. Get over it. The focus here is on
161 * supporting multi-way replication on PUT, with minimal complexity. These
162 * were the two easiest libraries to use, and they both want to allocate
163 * their own buffers so we're kind of stuck with the copies unless we want
164 * to buffer whole files in memory (which would have to be virtual since
165 * they're potentialy bigger than physical) or explicitly ping them through
166 * a local filesystem. We could potentially take over scheduling from one
167 * or both to avoid some of the context switching, but those interfaces are
168 * significantly more error-prone and (at least in CURL's case) buggy.
169 *
170 * For a GET, the CURL child acts as a producer while the MHD parent acts
171 * as consumer. For a PUT, the MHD parent is the producer and the CURL
172 * child is the consumer. For GET the MHD component is invoked via a
173 * callback set up in the access handler; for PUT it's invoked via repeated
174 * calls to the access handler itself. Either way, the producer's job is
175 * to post its pointer+length to the my_state structure and then wait for
176 * all consumers to check back in saying they've finished it. This might
177 * involve multiple passes through each consumer for one pass through the
178 * single producer. When the producer is done, it does a similar handshake
179 * with the consumers. Each consumer has its own pipe_private structure,
180 * containing a pointer to the shared my_state plus a per-consumer offset
181 * into the current chunk.
182 *
183 * Attribute functions don't use CURL, so they do much simpler in-memory
184 * buffering. Queries also don't use CURL, but the MHD POST interface
185 * introduces some of its own complexity so see below for that.
186 **********/
187
188 void
189 simple_closer (void *ctx)
190 {
191 my_state *ms = ctx;
192
193 DPRINTF("%s: cleaning up\n",__func__);
194 free_ms(ms);
195 }
196
197 void
198 child_closer (void * ctx)
199 {
200 pipe_private *pp = ctx;
201
202 DPRINTF("in %s\n",__func__);
203
204 free(pp);
205 }
206
207 /* Invoked from MHD. */
208 int
209 proxy_get_cons (void *ctx, uint64_t pos, char *buf, int max)
210 {
211 pipe_private *pp = ctx;
212 pipe_shared *ps = pp->shared;
213 my_state *ms = ps->owner;
214 int done;
215 void *child_res;
216
217 (void)pos;
218
219 DPRINTF("consumer asked to read %d\n",max);
220
221 if (pipe_cons_wait(pp)) {
222 DPRINTF("consumer offset %zu into %zu\n",
223 pp->offset, ps->data_len);
224 done = ps->data_len - pp->offset;
225 if (done > max) {
226 done = max;
227 }
228 memcpy(buf,ps->data_ptr+pp->offset,done);
229 pp->offset += done;
230 DPRINTF("consumer copied %d, new offset %zu\n",
231 done, pp->offset);
232 if (pp->offset == ps->data_len) {
233 DPRINTF("consumer finished chunk\n");
234 pipe_cons_signal(pp, 0);
235 }
236 }
237 else {
238 done = -1;
239 }
240
241 if (done == (-1)) {
242 child_res = NULL;
243 pthread_join(ms->backend_th,&child_res);
244 if (child_res == THREAD_FAILED) {
245 ms->rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
246 }
247 if (ms->from_master) {
248 pthread_join(ms->cache_th,NULL);
249 /* TBD: do something about cache failure? */
250 }
251 free_ms(ms);
252 }
253
254 return done;
255 }
256
257 int
258 proxy_get_data (void *cctx, struct MHD_Connection *conn, const char *url,
259 const char *method, const char *version, const char *data,
260 size_t *data_size, void **rctx)
261 {
262 struct MHD_Response *resp;
263 my_state *ms = *rctx;
264 pipe_private *pp;
265 pipe_private *pp2;
266 char *my_etag;
267 const char *user_etag;
268
269 (void)cctx;
270 (void)method;
271 (void)version;
272 (void)data;
273 (void)data_size;
274
275 DPRINTF("PROXY GET DATA %s\n",url);
276
277 ms->url = strdup(url);
278 if (!ms->url) {
279 return MHD_NO;
280 }
281 ms->cleanup |= CLEANUP_URL;
282
283 my_etag = meta_has_copy(ms->bucket,ms->key,me);
284 if (my_etag) {
285 user_etag = MHD_lookup_connection_value(
286 conn, MHD_HEADER_KIND, "If-None-Match");
287 if (user_etag && !strcmp(user_etag,my_etag)) {
288 DPRINTF("ETag match!\n");
289 free(my_etag);
290 resp = MHD_create_response_from_data(0,NULL,
291 MHD_NO,MHD_NO);
292 MHD_queue_response(conn,MHD_HTTP_NOT_MODIFIED,resp);
293 MHD_destroy_response(resp);
294 return MHD_YES;
295 }
296 free(my_etag);
297 ms->from_master = 0;
298 }
299 else {
300 DPRINTF("%s/%s not found locally\n",ms->bucket,ms->key);
301 if (!master_host) {
302 DPRINTF(" that means it doesn't exist\n");
303 resp = MHD_create_response_from_data(0,NULL,
304 MHD_NO,MHD_NO);
305 MHD_queue_response(conn,MHD_HTTP_NOT_FOUND,resp);
306 MHD_destroy_response(resp);
307 return MHD_YES;
308 }
309 DPRINTF(" will fetch from %s:%u\n", master_host,master_port);
310 ms->from_master = 1;
311 }
312
313 pipe_init_shared(&ms->pipe,ms,ms->from_master+1);
314 pp = pipe_init_private(&ms->pipe);
315 if (!pp) {
316 return MHD_NO;
317 }
318 /* Master is always assumed to be CURL (i.e. our own protocol) */
319 if (ms->from_master) {
320 pthread_create(&ms->backend_th,NULL,
321 curl_func_tbl.get_child_func,ms);
322 }
323 else {
324 pthread_create(&ms->backend_th,NULL,
325 main_func_tbl->get_child_func,ms);
326 }
327 /* TBD: check return value */
328
329 if (ms->from_master) {
330 pp2 = pipe_init_private(&ms->pipe);
331 if (!pp2) {
332 return MHD_NO;
333 }
334 pthread_create(&ms->cache_th,NULL,
335 main_func_tbl->cache_child_func,pp2);
336 /* TBD: check return value */
337 }
338 else {
339 pp2 = NULL;
340 }
341
342 resp = MHD_create_response_from_callback(
343 MHD_SIZE_UNKNOWN, 65536, proxy_get_cons, pp, child_closer);
344 if (!resp) {
345 fprintf(stderr,"MHD_crfc failed\n");
346 if (pp2) {
347 /* TBD: terminate thread */
348 free(pp2);
349 }
350 child_closer(pp);
351 return MHD_NO;
352 }
353 MHD_queue_response(conn,ms->rc,resp);
354 MHD_destroy_response(resp);
355
356 return MHD_YES;
357 }
358
359 void
360 recheck_replication (my_state * ms, char *policy)
361 {
362 int rc;
363 int free_it = FALSE;
364 char fixed[MAX_FIELD_LEN];
365
366 if (is_reserved(ms->key,reserved_name)) {
367 DPRINTF("declining to replicate reserved object %s\n",ms->key);
368 return;
369 }
370
371 if (!policy && ms->dict) {
372 DPRINTF("using new policy for %s/%s\n",ms->bucket,ms->key);
373 policy = g_hash_table_lookup(ms->dict,"_policy");
374 }
375
376 if (!policy) {
377 /* If we get a policy here or below, we have to free it. */
378 free_it = TRUE;
379 DPRINTF("fetching policy for %s/%s\n",ms->bucket,ms->key);
380 rc = meta_get_value(ms->bucket,ms->key, "_policy", &policy);
381 }
382
383 if (!policy) {
384 DPRINTF(" inheriting policy from %s\n",ms->bucket);
385 rc = meta_get_value(ms->bucket,
386 "_default", "_policy", &policy);
387 }
388
389 if (policy) {
390 DPRINTF(" implementing policy %s\n",policy);
391 /*
392 * Can't use ms->url here because it might be a bucket POST
393 * and in that case ms->url points to the bucket.
394 */
395 snprintf(fixed,sizeof(fixed),"%s/%s",ms->bucket,ms->key);
396 replicate(fixed,0,policy);
397 if (free_it) {
398 free(policy);
399 }
400 }
401 else {
402 DPRINTF(" could not find a policy anywhere!\n");
403 }
404 }
405
406 int
407 proxy_put_data (void *cctx, struct MHD_Connection *conn, const char *url,
408 const char *method, const char *version, const char *data,
409 size_t *data_size, void **rctx)
410 {
411 struct MHD_Response *resp;
412 my_state *ms = *rctx;
413 pipe_private *pp;
414 int rc;
415 char *etag = NULL;
416 void *child_res;
417
418 (void)cctx;
419 (void)method;
420 (void)version;
421
422 DPRINTF("PROXY PUT DATA %s (%zu)\n",url,*data_size);
423
424 if (ms->state == MS_NEW) {
425 if (!validate_put(conn) || !validate_url(url)) {
426 DPRINTF("rejecting %s\n",url);
427 resp = MHD_create_response_from_data(0,NULL,
428 MHD_NO,MHD_NO);
429 if (!resp) {
430 return MHD_NO;
431 }
432 MHD_queue_response(conn,MHD_HTTP_FORBIDDEN,resp);
433 MHD_destroy_response(resp);
434 return MHD_YES;
435 }
436 ms->state = MS_NORMAL;
437 ms->url = strdup(url);
438 if (!ms->url) {
439 return MHD_NO;
440 }
441 ms->cleanup |= CLEANUP_URL;
442 ms->size = 0;
443 pipe_init_shared(&ms->pipe,ms,1);
444 pp = pipe_init_private(&ms->pipe);
445 if (!pp) {
446 return MHD_NO;
447 }
448 pthread_create(&ms->backend_th,NULL,
449 main_func_tbl->put_child_func,pp);
450 /* TBD: check return value */
451
452 /*
453 * Do the initial handshake with children. If we return from
454 * this callback without an error response, Microhttpd posts
455 * the "100 Continue" header and the client starts sending
456 * the data. We must report errors here or forever keep
457 * out peace.
458 */
459 rc = pipe_prod_wait_init(&ms->pipe);
460 if (rc < 0) {
461 DPRINTF("producer wait failed\n");
462 resp = MHD_create_response_from_data(0,NULL,
463 MHD_NO,MHD_NO);
464 if (!resp) {
465 return MHD_NO;
466 }
467 MHD_queue_response(conn,MHD_HTTP_INTERNAL_SERVER_ERROR,
468 resp);
469 MHD_destroy_response(resp);
470 } else if (rc > 0) {
471 /*
472 * Note that we fail here even if 1 of N replicas fail.
473 * Might want to fix this when we start looping over
474 * pipe_init_private() above.
475 */
476 DPRINTF("producer replicas failed (%u of %u)\n",
477 rc, ms->pipe.cons_total);
478 resp = MHD_create_response_from_data(0,NULL,
479 MHD_NO,MHD_NO);
480 if (!resp) {
481 return MHD_NO;
482 }
483 MHD_queue_response(conn,MHD_HTTP_INTERNAL_SERVER_ERROR,
484 resp);
485 MHD_destroy_response(resp);
486 } else {
487 DPRINTF("producer proceeding\n");
488 }
489 }
490 else if (*data_size) {
491 pipe_prod_signal(&ms->pipe,(void *)data,*data_size);
492 ms->size += *data_size;
493 DPRINTF("producer chunk finished\n");
494 *data_size = 0;
495 }
496 else {
497 pipe_prod_finish(&ms->pipe);
498 pthread_join(ms->backend_th,&child_res);
499 if (child_res == THREAD_FAILED) {
500 DPRINTF("thread failed\n");
501 rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
502 }
503 else if (ms->pipe.cons_error == ms->pipe.cons_total) {
504 DPRINTF("all %u consumers failed\n",
505 ms->pipe.cons_error);
506 rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
507 }
508 else {
509 if (master_host) {
510 meta_got_copy(ms->bucket,ms->key,me);
511 etag = NULL;
512 }
513 else {
514 etag = meta_did_put(ms->bucket,ms->key,me,
515 ms->size);
516 }
517 DPRINTF("rereplicate (obj PUT)\n");
518 recheck_replication(ms,NULL);
519 rc = MHD_HTTP_OK;
520 }
521 free_ms(ms);
522 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
523 if (!resp) {
524 free(etag);
525 return MHD_NO;
526 }
527 if (etag) {
528 MHD_add_response_header(resp,"ETag",etag);
529 free(etag);
530 }
531 MHD_queue_response(conn,rc,resp);
532 MHD_destroy_response(resp);
533 }
534
535 return MHD_YES;
536 }
537
538 int
539 proxy_get_attr (void *cctx, struct MHD_Connection *conn, const char *url,
540 const char *method, const char *version, const char *data,
541 size_t *data_size, void **rctx)
542 {
543 struct MHD_Response *resp;
544 char *fixed;
545 my_state *ms = *rctx;
546 int rc = MHD_HTTP_NOT_FOUND;
547
548 (void)cctx;
549 (void)method;
550 (void)version;
551 (void)data;
552 (void)data_size;
553
554 DPRINTF("PROXY GET ATTR %s\n",url);
555
556 if (meta_get_value(ms->bucket,ms->key,ms->attr,&fixed) == 0) {
557 resp = MHD_create_response_from_data(strlen(fixed),fixed,
558 MHD_YES,MHD_NO);
559 rc = MHD_HTTP_OK;
560 }
561 else {
562 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
563 }
564 if (!resp) {
565 return MHD_NO;
566 }
567 MHD_queue_response(conn,rc,resp);
568 MHD_destroy_response(resp);
569
570 return MHD_YES;
571 }
572
573 int
574 proxy_put_attr (void *cctx, struct MHD_Connection *conn, const char *url,
575 const char *method, const char *version, const char *data,
576 size_t *data_size, void **rctx)
577 {
578 struct MHD_Response *resp;
579 my_state *ms = *rctx;
580 const char *attrval;
581 int send_resp = 0;
582
583 (void)cctx;
584 (void)method;
585 (void)version;
586
587 DPRINTF("PROXY PUT ATTR %s (%zu)\n",url,*data_size);
588
589 if (ms->state == MS_NEW) {
590 ms->state = MS_NORMAL;
591 ms->url = strdup(url);
592 if (!ms->url) {
593 return MHD_NO;
594 }
595 ms->cleanup |= CLEANUP_URL;
596 attrval = MHD_lookup_connection_value(conn,MHD_HEADER_KIND,
597 "X-redhat-value");
598 if (attrval) {
599 meta_set_value(ms->bucket,ms->key,ms->attr,
600 (char *)attrval);
601 send_resp = 1;
602 }
603 }
604 else if (*data_size) {
605 if (ms->pipe.data_len) {
606 ms->pipe.data_len += *data_size;
607 char *p = realloc(ms->pipe.data_ptr,ms->pipe.data_len);
608 if (!p) {
609 return MHD_NO;
610 }
611 ms->pipe.data_ptr = p;
612 }
613 else {
614 ms->pipe.data_len = *data_size + 1;
615 ms->pipe.data_ptr = malloc(ms->pipe.data_len);
616 if (!ms->pipe.data_ptr) {
617 return MHD_NO;
618 }
619 ((char *)ms->pipe.data_ptr)[0] = '\0';
620 ms->cleanup |= CLEANUP_BUF_PTR;
621 }
622 (void)strncat(ms->pipe.data_ptr,data,*data_size);
623 /* TBD: check return value */
624 *data_size = 0;
625 }
626 else {
627 if (!ms->pipe.data_ptr) {
628 return MHD_NO;
629 }
630 if (is_reserved(ms->attr,reserved_attr)) {
631 resp = MHD_create_response_from_data(
632 0,NULL,MHD_NO,MHD_NO);
633 if (!resp) {
634 return MHD_NO;
635 }
636 MHD_queue_response(conn,MHD_HTTP_BAD_REQUEST,
637 resp);
638 MHD_destroy_response(resp);
639 free_ms(ms);
640 return MHD_YES;
641 }
642 meta_set_value(ms->bucket,ms->key,ms->attr,ms->pipe.data_ptr);
643 /*
644 * We should always re-replicate, because the replication
645 * policy might refer to this attr.
646 */
647 DPRINTF("rereplicate (attr PUT)\n");
648 recheck_replication(ms,NULL);
649 free_ms(ms);
650 send_resp = 1;
651 }
652
653 if (send_resp) {
654 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
655 if (!resp) {
656 return MHD_NO;
657 }
658 MHD_queue_response(conn,MHD_HTTP_CREATED,resp);
659 MHD_destroy_response(resp);
660 /*
661 * TBD: check if the attribute was a replication policy, and
662 * start/stop replication activities as appropriate.
663 */
664 }
665
666 return MHD_YES;
667 }
668
669 /**********
670 * For queries, we have to deal with MHD's post-iterator interface (not
671 * quite the same as the data-iteration even though we use it that way) on
672 * one side, and a query-iterator interface on the other. Data on both
673 * sides could be quite large, so we can't just stick them in header lines.
674 * We do still buffer the query in memory, though. Once that's done, we do
675 * very simple parsing - it will be more complicated later - and create the
676 * query iterator. That's also driven by MHD, this time though the
677 * content-callback interface, and repeatedly calls in to the metadata
678 * module to fetch one object name at a time.
679 **********/
680
681 int
682 query_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
683 const char *filename, const char *content_type,
684 const char *transfer_encoding, const char *data,
685 uint64_t off, size_t size)
686 {
687 (void)ctx;
688 (void)kind;
689 (void)key;
690 (void)filename;
691 (void)content_type;
692 (void)transfer_encoding;
693 (void)data;
694 (void)off;
695 (void)size;
696
697 /* We actually accumulate the data in proxy_query. */
698 return MHD_YES;
699 }
700
701 /* MHD reader function during queries. Return -1 for EOF. */
702 int
703 proxy_query_func (void *ctx, uint64_t pos, char *buf, int max)
704 {
705 my_state *ms = ctx;
706 int len;
707 const char *accept_hdr;
708 char *bucket;
709 char *key;
710
711 (void)pos;
712
713 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
714 "Accept");
715
716 if (!ms->gen_ctx) {
717 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
718 if (!ms->gen_ctx) {
719 return -1;
720 }
721 ms->cleanup |= CLEANUP_TMPL;
722 len = tmpl_obj_header(ms->gen_ctx);
723 if (!len) {
724 return -1;
725 }
726 if (len > max) {
727 len = max;
728 }
729 memcpy(buf,ms->gen_ctx->buf,len);
730 return len;
731 }
732
733 if (ms->gen_ctx == TMPL_CTX_DONE) {
734 return -1;
735 }
736
737 for(;;) {
738 if (!meta_query_next(ms->query,&bucket,&key)) {
739 break;
740 }
741 if (is_reserved(key,reserved_name)) {
742 continue;
743 }
744 len = tmpl_obj_entry(ms->gen_ctx,bucket,key);
745 if (!len) {
746 return -1;
747 }
748 if (len > max) {
749 len = max;
750 }
751 memcpy(buf,ms->gen_ctx->buf,len);
752 return len;
753 }
754
755 len = tmpl_obj_footer(ms->gen_ctx);
756 if (!len) {
757 return -1;
758 }
759 if (len > max) {
760 len = max;
761 }
762 memcpy(buf,ms->gen_ctx->buf,len);
763 free(ms->gen_ctx);
764 ms->cleanup &= ~CLEANUP_TMPL;
765 ms->gen_ctx = TMPL_CTX_DONE;
766 return len;
767 }
768
769 int
770 proxy_query (void *cctx, struct MHD_Connection *conn, const char *url,
771 const char *method, const char *version, const char *data,
772 size_t *data_size, void **rctx)
773 {
774 struct MHD_Response *resp;
775 my_state *ms = *rctx;
776
777 (void)cctx;
778 (void)method;
779 (void)version;
780
781 DPRINTF("PROXY QUERY %s (%zu)\n",url,*data_size);
782
783 if (ms->state == MS_NEW) {
784 ms->state = MS_NORMAL;
785 ms->post = MHD_create_post_processor(conn,4096,
786 query_iterator,ms);
787 ms->cleanup |= CLEANUP_POST;
788 }
789 else if (*data_size) {
790 MHD_post_process(ms->post,data,*data_size);
791 if (ms->pipe.data_len) {
792 ms->pipe.data_len += *data_size;
793 char *p = realloc(ms->pipe.data_ptr,ms->pipe.data_len);
794 if (!p) {
795 return MHD_NO;
796 }
797 ms->pipe.data_ptr = p;
798 }
799 else {
800 ms->pipe.data_len = *data_size + 1;
801 ms->pipe.data_ptr = malloc(ms->pipe.data_len);
802 if (!ms->pipe.data_ptr) {
803 return MHD_NO;
804 }
805 ((char *)ms->pipe.data_ptr)[0] = '\0';
806 ms->cleanup |= CLEANUP_BUF_PTR;
807 }
808 (void)strncat(ms->pipe.data_ptr,data,*data_size);
809 /* TBD: check return value */
810 *data_size = 0;
811 }
812 else {
813 if (!ms->pipe.data_ptr) {
814 return MHD_NO;
815 }
816 ms->query = meta_query_new(ms->bucket,NULL,ms->pipe.data_ptr);
817 ms->cleanup |= CLEANUP_QUERY;
818 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
819 65536, proxy_query_func, ms, simple_closer);
820 if (!resp) {
821 fprintf(stderr,"MHD_crfc failed\n");
822 simple_closer(ms);
823 return MHD_NO;
824 }
825 MHD_add_response_header(resp,"Content-Type","text/xml");
826 MHD_queue_response(conn,MHD_HTTP_OK,resp);
827 MHD_destroy_response(resp);
828 }
829
830 return MHD_YES;
831 }
832
833 int
834 proxy_list_objs (void *cctx, struct MHD_Connection *conn, const char *url,
835 const char *method, const char *version, const char *data,
836 size_t *data_size, void **rctx)
837 {
838 my_state *ms = *rctx;
839 struct MHD_Response *resp;
840
841 (void)cctx;
842 (void)url;
843 (void)method;
844 (void)version;
845 (void)data;
846 (void)data_size;
847
848 ms->query = meta_query_new((char *)ms->bucket,NULL,NULL);
849 ms->cleanup |= CLEANUP_QUERY;
850
851 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
852 65536, proxy_query_func, ms, simple_closer);
853 if (!resp) {
854 fprintf(stderr,"MHD_crfc failed\n");
855 simple_closer(ms);
856 return MHD_NO;
857 }
858
859 MHD_queue_response(conn,MHD_HTTP_OK,resp);
860 MHD_destroy_response(resp);
861 return MHD_YES;
862 }
863
864 int
865 proxy_delete (void *cctx, struct MHD_Connection *conn, const char *url,
866 const char *method, const char *version, const char *data,
867 size_t *data_size, void **rctx)
868 {
869 my_state *ms = *rctx;
870 struct MHD_Response *resp;
871 char *copied_url;
872 char *bucket;
873 char *key;
874 char *stctx = NULL;
875 int rc;
876
877 (void)cctx;
878 (void)method;
879 (void)version;
880 (void)data;
881 (void)data_size;
882
883 DPRINTF("PROXY DELETE %s\n",url);
884
885 rc = main_func_tbl->delete_func(ms->bucket,ms->key,(char *)url);
886 if (rc != MHD_YES) {
887 return rc;
888 }
889
890 copied_url = strdup(url);
891 assert (copied_url);
892 bucket = strtok_r(copied_url,"/",&stctx);
893 key = strtok_r(NULL,"/",&stctx);
894 meta_delete(bucket,key);
895 free(copied_url);
896
897 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
898 if (!resp) {
899 return MHD_NO;
900 }
901 MHD_add_response_header(resp,"Content-Type","text/xml");
902 MHD_queue_response(conn,MHD_HTTP_OK,resp);
903 MHD_destroy_response(resp);
904
905 replicate_delete((char *)url);
906 return MHD_YES;
907 }
908
909 /* TBD: get actual bucket list */
910 typedef struct {
911 char *rel;
912 char *link;
913 } fake_bucket_t;
914
915 fake_bucket_t fake_bucket_list[] = {
916 { "bucket_factory", "_new" },
917 { "provider_list", "_providers" },
918 };
919
920 int
921 root_blob_generator (void *ctx, uint64_t pos, char *buf, int max)
922 {
923 my_state *ms = ctx;
924 fake_bucket_t * fb;
925 int len;
926 const char *accept_hdr;
927 const char *host;
928 char *bucket;
929 char *key;
930
931 (void)pos;
932
933 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
934 "Accept");
935 host = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,"Host");
936
937 if (!ms->gen_ctx) {
938 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
939 if (!ms->gen_ctx) {
940 return -1;
941 }
942 ms->cleanup |= CLEANUP_TMPL;
943 ms->gen_ctx->base = host;
944 len = tmpl_root_header(ms->gen_ctx,"image_warehouse","1.0");
945 if (!len) {
946 return -1;
947 }
948 if (len > max) {
949 len = max;
950 }
951 memcpy(buf,ms->gen_ctx->buf,len);
952 return len;
953 }
954
955 if (ms->gen_ctx == TMPL_CTX_DONE) {
956 return -1;
957 }
958
959 if (ms->gen_ctx->index < ARRAY_SIZE(fake_bucket_list)) {
960 fb = fake_bucket_list + ms->gen_ctx->index;
961 len = tmpl_root_entry(ms->gen_ctx,fb->rel,fb->link);
962 if (!len) {
963 return -1;
964 }
965 if (len > max) {
966 len = max;
967 }
968 memcpy(buf,ms->gen_ctx->buf,len);
969 return len;
970 }
971
972 if (meta_query_next(ms->query,&bucket,&key)) {
973 len = tmpl_root_entry(ms->gen_ctx,"bucket",bucket);
974 if (!len) {
975 return -1;
976 }
977 if (len > max) {
978 len = max;
979 }
980 memcpy(buf,ms->gen_ctx->buf,len);
981 return len;
982 }
983
984 len = tmpl_root_footer(ms->gen_ctx);
985 if (!len) {
986 return -1;
987 }
988 if (len > max) {
989 len = max;
990 }
991 memcpy(buf,ms->gen_ctx->buf,len);
992 free(ms->gen_ctx);
993 ms->cleanup &= ~CLEANUP_TMPL;
994 ms->gen_ctx = TMPL_CTX_DONE;
995 return len;
996 }
997
998 int
999 proxy_api_root (void *cctx, struct MHD_Connection *conn, const char *url,
1000 const char *method, const char *version, const char *data,
1001 size_t *data_size, void **rctx)
1002 {
1003 struct MHD_Response *resp = NULL;
1004 unsigned int rc = MHD_HTTP_OK;
1005 my_state *ms = *rctx;
1006
1007 (void)cctx;
1008 (void)method;
1009 (void)version;
1010 (void)data;
1011
1012 DPRINTF("PROXY API ROOT (%s, %zu)\n",url,*data_size);
1013
1014 ms->query = meta_query_new(NULL,"_default",NULL);
1015 if (!ms->query) {
1016 free_ms(ms);
1017 return MHD_NO;
1018 }
1019 ms->cleanup |= CLEANUP_QUERY;
1020
1021 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
1022 65536, root_blob_generator, ms, simple_closer);
1023 if (!resp) {
1024 return MHD_NO;
1025 }
1026 MHD_add_response_header(resp,"Content-Type","text/xml");
1027 MHD_queue_response(conn,rc,resp);
1028 MHD_destroy_response(resp);
1029
1030 return MHD_YES;
1031
1032 }
1033
1034 int
1035 post_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
1036 const char *filename, const char *content_type,
1037 const char *transfer_encoding, const char *data,
1038 uint64_t off, size_t size)
1039 {
1040 char *old_val;
1041 size_t old_len;
1042 char *new_val;
1043
1044 (void)kind;
1045 (void)filename;
1046 (void)content_type;
1047 (void)transfer_encoding;
1048 (void)off;
1049
1050 printf("adding %s, size=%zu\n",key,size);
1051
1052 // TBD: don't assume that values are null-terminated strings
1053 old_val = g_hash_table_lookup(ctx,key);
1054 if (old_val) {
1055 old_len = strlen(old_val);
1056 new_val = malloc(old_len+size+1);
1057 if (!new_val) {
1058 return MHD_NO;
1059 }
1060 memcpy(new_val,old_val,old_len);
1061 memcpy(new_val+old_len,data,size);
1062 new_val[old_len+size] = '\0';
1063 }
1064 else {
1065 new_val = malloc(size+1);
1066 if (!new_val) {
1067 return MHD_NO;
1068 }
1069 memcpy(new_val,data,size);
1070 new_val[size] = '\0';
1071 }
1072
1073 g_hash_table_insert(ctx,strdup(key),new_val);
1074 /* TBD: check return value for strdups (none avail for insert) */
1075 return MHD_YES;
1076 }
1077
1078 /* Returns TRUE if we found an *invalid* key. */
1079 gboolean
1080 post_find (gpointer key, gpointer value, gpointer ctx)
1081 {
1082 (void)value;
1083 (void)ctx;
1084
1085 if (!is_reserved(key,reserved_attr)) {
1086 return FALSE;
1087 }
1088
1089 DPRINTF("bad attr %s\n", (char *)key);
1090 return TRUE;
1091 }
1092
1093 void
1094 post_foreach (gpointer key, gpointer value, gpointer ctx)
1095 {
1096 my_state *ms = ctx;
1097
1098 DPRINTF("setting %s = %s for %s/%s\n",(char *)key, (char *)value,
1099 ms->bucket,ms->key);
1100 meta_set_value(ms->bucket,ms->key,key,value);
1101 }
1102
1103 int
1104 create_bucket (char *name)
1105 {
1106 int rc;
1107
1108 if (is_reserved(name,reserved_name)) {
1109 return MHD_HTTP_BAD_REQUEST;
1110 }
1111
1112 rc = main_func_tbl->bcreate_func(name);
1113 if (rc == MHD_HTTP_OK) {
1114 if (meta_set_value(name,"_default", "_policy","0") != 0) {
1115 DPRINTF("default-policy " "create failed\n");
1116 /* Non-fatal. */
1117 }
1118 DPRINTF("created bucket %s\n",name);
1119 /*
1120 * There's not a whole lot to do about bucket-creation
1121 * failures on replicas, other than to report them, unless
1122 * we adopt an "all or nothing" approach and unwind the
1123 * create on the primary as well. Then what if that fails?
1124 * It's just one example of the general "fewer replicas
1125 * than desired" distributed-system problem, not worth a
1126 * point solution here/now. Revisit when we have a more
1127 * general replica-repair policy/system in place.
1128 */
1129 replicate_bcreate(name);
1130 }
1131
1132 return rc;
1133 }
1134
1135 int
1136 proxy_bucket_post (void *cctx, struct MHD_Connection *conn, const char *url,
1137 const char *method, const char *version, const char *data,
1138 size_t *data_size, void **rctx)
1139 {
1140 struct MHD_Response *resp;
1141 my_state *ms = *rctx;
1142 int rc;
1143 char *key;
1144
1145 (void)cctx;
1146 (void)method;
1147 (void)version;
1148
1149 DPRINTF("PROXY POST (%s, %zu)\n",url,*data_size);
1150
1151 if (ms->state == MS_NEW) {
1152 ms->state = MS_NORMAL;
1153 ms->url = (char *)url;
1154 ms->dict = g_hash_table_new_full(
1155 g_str_hash,g_str_equal,free,free);
1156 ms->cleanup |= CLEANUP_DICT;
1157 ms->post = MHD_create_post_processor(conn,4096,
1158 post_iterator,ms->dict);
1159 ms->cleanup |= CLEANUP_POST;
1160 }
1161 else if (*data_size) {
1162 MHD_post_process(ms->post,data,*data_size);
1163 *data_size = 0;
1164 }
1165 else {
1166 rc = MHD_HTTP_BAD_REQUEST;
1167 key = g_hash_table_lookup(ms->dict,"key");
1168 if (key) {
1169 strncpy(ms->key,key,MAX_FIELD_LEN-1);
1170 g_hash_table_remove(ms->dict,"key");
1171 if (!g_hash_table_find(ms->dict,post_find,ms)) {
1172 g_hash_table_foreach(ms->dict,post_foreach,ms);
1173 DPRINTF("rereplicate (bucket POST)\n");
1174 recheck_replication(ms,NULL);
1175 rc = MHD_HTTP_OK;
1176 }
1177 }
1178 else if (!strcmp(ms->bucket,"_new")) {
1179 key = g_hash_table_lookup(ms->dict,"name");
1180 if (key != NULL) {
1181 rc = create_bucket(key);
1182 }
1183 }
1184 else {
1185 DPRINTF("A parameter is MISSING (fail)\n");
1186 }
1187 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1188 if (!resp) {
1189 fprintf(stderr,"MHD_crfd failed\n");
1190 return MHD_NO;
1191 }
1192 MHD_queue_response(conn,rc,resp);
1193 MHD_destroy_response(resp);
1194 }
1195
1196 return MHD_YES;
1197 }
1198
1199 int
1200 check_location (my_state *ms)
1201 {
1202 char *loc = g_hash_table_lookup(ms->dict,"depot");
1203
1204 if (!loc) {
1205 DPRINTF("missing loc on check for %s/%s\n",ms->bucket,ms->key);
1206 return MHD_HTTP_BAD_REQUEST;
1207 }
1208
1209 if (!meta_has_copy(ms->bucket,ms->key,loc)) {
1210 DPRINTF("did not find %s/%s at %s\n",ms->bucket,ms->key,loc);
1211 return MHD_HTTP_NOT_FOUND;
1212 }
1213
1214 /* TBD: meta_has_copy returns an etag which we should check */
1215 DPRINTF("found %s/%s at %s\n",ms->bucket,ms->key,loc);
1216 return MHD_HTTP_OK;
1217 }
1218
1219 int
1220 register_image (my_state *ms)
1221 {
1222 char *site;
1223 provider_t prov;
1224 int i;
1225 char *next;
1226
1227 site = g_hash_table_lookup(ms->dict,"site");
1228 if (!site) {
1229 printf("site MISSING\n");
1230 return MHD_HTTP_BAD_REQUEST;
1231 }
1232
1233 next = index(site,':');
1234 if (next) {
1235 *(next++) = '\0';
1236 }
1237
1238 for (i = 0; get_provider(i,&prov); ++i) {
1239 if (strcmp(prov.name,site)) {
1240 continue;
1241 }
1242 return prov.func_tbl->register_func(ms,&prov,next,ms->dict);
1243 }
1244
1245 DPRINTF("site %s not found\n",site);
1246 return MHD_HTTP_BAD_REQUEST;
1247 }
1248
1249 int
1250 proxy_object_post (void *cctx, struct MHD_Connection *conn, const char *url,
1251 const char *method, const char *version, const char *data,
1252 size_t *data_size, void **rctx)
1253 {
1254 struct MHD_Response *resp;
1255 my_state *ms = *rctx;
1256 int rc;
1257 char *op;
1258
1259 (void)cctx;
1260 (void)method;
1261 (void)version;
1262
1263 DPRINTF("PROXY POST (%s, %zu)\n",url,*data_size);
1264
1265 if (ms->state == MS_NEW) {
1266 ms->state = MS_NORMAL;
1267 ms->url = (char *)url;
1268 ms->dict = g_hash_table_new_full(
1269 g_str_hash,g_str_equal,free,free);
1270 ms->cleanup |= CLEANUP_DICT;
1271 ms->post = MHD_create_post_processor(conn,4096,
1272 post_iterator,ms->dict);
1273 ms->cleanup |= CLEANUP_POST;
1274 }
1275 else if (*data_size) {
1276 MHD_post_process(ms->post,data,*data_size);
1277 *data_size = 0;
1278 }
1279 else {
1280 rc = MHD_HTTP_BAD_REQUEST;
1281 if (!g_hash_table_find(ms->dict,post_find,ms)) {
1282 op = g_hash_table_lookup(ms->dict,"op");
1283 if (op) {
1284 if (!strcmp(op,"push")) {
1285 DPRINTF("rereplicate (obj POST)\n");
1286 recheck_replication(ms,NULL);
1287 rc = MHD_HTTP_OK;
1288 }
1289 else if (!strcmp(op,"check")) {
1290 rc = check_location(ms);
1291 }
1292 else if (!strcmp(op,"register")) {
1293 rc = register_image(ms);
1294 }
1295 else {
1296 DPRINTF("unknown op %s for %s/%s\n",
1297 op, ms->bucket, ms->key);
1298 }
1299 }
1300 else {
1301 DPRINTF("op is MISSING (fail)\n");
1302 }
1303 }
1304 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1305 if (!resp) {
1306 fprintf(stderr,"MHD_crfd failed\n");
1307 return MHD_NO;
1308 }
1309 MHD_queue_response(conn,rc,resp);
1310 MHD_destroy_response(resp);
1311 }
1312
1313 return MHD_YES;
1314
1315 }
1316
1317
1318 int
1319 prov_list_generator (void *ctx, uint64_t pos, char *buf, int max)
1320 {
1321 my_state *ms = ctx;
1322 int len;
1323 provider_t prov;
1324 const char *accept_hdr;
1325
1326 (void)pos;
1327
1328 accept_hdr = MHD_lookup_connection_value(ms->conn,MHD_HEADER_KIND,
1329 "Accept");
1330
1331 if (!ms->gen_ctx) {
1332 ms->gen_ctx = tmpl_get_ctx(accept_hdr);
1333 if (!ms->gen_ctx) {
1334 return -1;
1335 }
1336 ms->cleanup |= CLEANUP_TMPL;
1337 len = tmpl_prov_header(ms->gen_ctx);
1338 if (!len) {
1339 return -1;
1340 }
1341 if (len > max) {
1342 len = max;
1343 }
1344 memcpy(buf,ms->gen_ctx->buf,len);
1345 return len;
1346 }
1347
1348 if (ms->gen_ctx == TMPL_CTX_DONE) {
1349 return -1;
1350 }
1351
1352 if (get_provider(ms->gen_ctx->index,&prov)) {
1353 len = tmpl_prov_entry(ms->gen_ctx,prov.name,prov.type,
1354 prov.host, prov.port, prov.username, prov.password);
1355 if (!len) {
1356 return -1;
1357 }
1358 if (len > max) {
1359 len = max;
1360 }
1361 memcpy(buf,ms->gen_ctx->buf,len);
1362 return len;
1363 }
1364
1365 len = tmpl_prov_footer(ms->gen_ctx);
1366 if (!len) {
1367 return -1;
1368 }
1369 if (len > max) {
1370 len = max;
1371 }
1372 memcpy(buf,ms->gen_ctx->buf,len);
1373 free(ms->gen_ctx);
1374 ms->cleanup &= ~CLEANUP_TMPL;
1375 ms->gen_ctx = TMPL_CTX_DONE;
1376 return len;
1377 }
1378
1379 int
1380 proxy_list_provs (void *cctx, struct MHD_Connection *conn, const char *url,
1381 const char *method, const char *version, const char *data,
1382 size_t *data_size, void **rctx)
1383 {
1384 struct MHD_Response *resp;
1385 my_state *ms = *rctx;
1386
1387 (void)cctx;
1388 (void)url;
1389 (void)method;
1390 (void)version;
1391 (void)data;
1392 (void)data_size;
1393
1394 resp = MHD_create_response_from_callback(MHD_SIZE_UNKNOWN,
1395 65536, prov_list_generator, ms, simple_closer);
1396 if (!resp) {
1397 fprintf(stderr,"MHD_crfd failed\n");
1398 simple_closer(ms);
1399 return MHD_NO;
1400 }
1401 MHD_queue_response(conn,MHD_HTTP_OK,resp);
1402 MHD_destroy_response(resp);
1403
1404 return MHD_YES;
1405 }
1406
1407 int
1408 prov_iterator (void *ctx, enum MHD_ValueKind kind, const char *key,
1409 const char *filename, const char *content_type,
1410 const char *transfer_encoding, const char *data,
1411 uint64_t off, size_t size)
1412 {
1413 (void)kind;
1414 (void)filename;
1415 (void)content_type;
1416 (void)transfer_encoding;
1417 (void)off;
1418
1419 g_hash_table_insert(ctx,strdup(key),strndup(data,size));
1420 /* TBD: check return value for strdups (none avail for insert) */
1421 return MHD_YES;
1422 }
1423
1424
1425 int
1426 proxy_update_prov (void *cctx, struct MHD_Connection *conn, const char *url,
1427 const char *method, const char *version, const char *data,
1428 size_t *data_size, void **rctx)
1429 {
1430 struct MHD_Response *resp;
1431 my_state *ms = *rctx;
1432 int rc;
1433 char *provider;
1434 char *username;
1435 char *password;
1436
1437 (void)cctx;
1438 (void)method;
1439 (void)version;
1440
1441 if (ms->state == MS_NEW) {
1442 ms->state = MS_NORMAL;
1443 ms->url = (char *)url;
1444 ms->dict = g_hash_table_new_full(
1445 g_str_hash,g_str_equal,free,free);
1446 ms->cleanup |= CLEANUP_DICT;
1447 ms->post = MHD_create_post_processor(conn,4096,
1448 prov_iterator,ms->dict);
1449 ms->cleanup |= CLEANUP_POST;
1450 }
1451 else if (*data_size) {
1452 MHD_post_process(ms->post,data,*data_size);
1453 *data_size = 0;
1454 }
1455 else {
1456 rc = MHD_HTTP_BAD_REQUEST;
1457 provider = g_hash_table_lookup(ms->dict,"provider");
1458 username = g_hash_table_lookup(ms->dict,"username");
1459 password = g_hash_table_lookup(ms->dict,"password");
1460 if (provider && username && password) {
1461 update_provider(provider,username,password);
1462 rc = MHD_HTTP_OK;
1463 }
1464 else {
1465 DPRINTF("provider/username/password MISSING\n");
1466 }
1467 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1468 if (!resp) {
1469 fprintf(stderr,"MHD_crfd failed\n");
1470 return MHD_NO;
1471 }
1472 MHD_queue_response(conn,rc,resp);
1473 MHD_destroy_response(resp);
1474 }
1475
1476 return MHD_YES;
1477 }
1478
1479 int
1480 proxy_create_bucket (void *cctx, struct MHD_Connection *conn, const char *url,
1481 const char *method, const char *version, const char *data,
1482 size_t *data_size, void **rctx)
1483 {
1484 struct MHD_Response *resp;
1485 my_state *ms = *rctx;
1486 int rc;
1487
1488 (void)cctx;
1489 (void)method;
1490 (void)version;
1491 (void)data;
1492 (void)data_size;
1493 (void)url;
1494
1495 /* curl -T moo.empty http://localhost:9090/_new by accident */
1496 rc = create_bucket(ms->bucket);
1497
1498 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1499 if (!resp) {
1500 fprintf(stderr,"MHD_crfd failed\n");
1501 return MHD_NO;
1502 }
1503 MHD_queue_response(conn,rc,resp);
1504 MHD_destroy_response(resp);
1505
1506 return MHD_YES;
1507 }
1508
1509 rule my_rules[] = {
1510 { /* get bucket list */
1511 "GET", URL_ROOT, proxy_api_root },
1512 { /* get object list */
1513 "GET", URL_BUCKET, proxy_list_objs },
1514 { /* create bucket */
1515 "PUT", URL_BUCKET, proxy_create_bucket },
1516 { /* get object data */
1517 "GET", URL_OBJECT, proxy_get_data },
1518 { /* get attribute data */
1519 "GET", URL_ATTR, proxy_get_attr },
1520 { /* put object data */
1521 "PUT", URL_OBJECT, proxy_put_data },
1522 { /* put attribute data */
1523 "PUT", URL_ATTR, proxy_put_attr },
1524 { /* create object and/or modify attributes */
1525 "POST", URL_BUCKET, proxy_bucket_post },
1526 { /* perform control operations on an object */
1527 "POST", URL_OBJECT, proxy_object_post },
1528 { /* query */
1529 "POST", URL_QUERY, proxy_query },
1530 { /* delete object */
1531 "DELETE", URL_OBJECT, proxy_delete },
1532 { /* delete attribute (TBD) */
1533 "DELETE", URL_ATTR, NULL },
1534 { /* get provider list */
1535 "GET", URL_PROVLIST, proxy_list_provs },
1536 { /* update a provider */
1537 "POST", URL_PROVLIST, proxy_update_prov },
1538 { NULL, 0, NULL }
1539 };
1540
1541 url_type
1542 parse_url (const char *url, my_state *ms)
1543 {
1544 unsigned short esize;
1545 unsigned short eindex;
1546 char *parts[URL_INVAL];
1547
1548 eindex = URL_ROOT;
1549 parts[URL_BUCKET] = ms->bucket;
1550 parts[URL_OBJECT] = ms->key;
1551 parts[URL_ATTR] = ms->attr;
1552
1553 for (;;) {
1554 while (*url == '/') {
1555 ++url;
1556 }
1557
1558 if (!*url) {
1559 if (eindex == URL_BUCKET) {
1560 if (!strcmp(ms->bucket,"_providers")) {
1561 eindex = URL_PROVLIST;
1562 }
1563 }
1564 else if (eindex == URL_OBJECT) {
1565 if (!strcmp(ms->key,"_query")) {
1566 eindex = URL_QUERY;
1567 }
1568 }
1569 break;
1570 }
1571
1572 if (++eindex >= URL_INVAL) {
1573 return URL_INVAL;
1574 }
1575 esize = 0;
1576
1577 while (*url && (*url != '/')) {
1578 parts[eindex][esize++] = *(url++);
1579 if (esize >= MAX_FIELD_LEN) {
1580 return URL_INVAL;
1581 }
1582 }
1583 }
1584
1585 return eindex;
1586 }
1587
1588 int
1589 access_handler (void *cctx, struct MHD_Connection *conn, const char *url,
1590 const char *method, const char *version, const char *data,
1591 size_t *data_size, void **rctx)
1592 {
1593 unsigned int i;
1594 url_type utype;
1595 struct MHD_Response *resp;
1596 my_state *ms = *rctx;
1597
1598 if (ms) {
1599 return ms->handler(cctx,conn,url,method,version,
1600 data,data_size,rctx);
1601 }
1602
1603 ms = malloc(sizeof(my_state));
1604 if (!ms) {
1605 return MHD_NO;
1606 }
1607 memset(ms,0,sizeof(*ms));
1608
1609 utype = parse_url(url,ms);
1610
1611 for (i = 0; my_rules[i].method; ++i) {
1612 if (utype != my_rules[i].utype) {
1613 continue;
1614 }
1615 if (strcmp(method,my_rules[i].method)) {
1616 continue;
1617 }
1618 if (!my_rules[i].handler) {
1619 break;
1620 }
1621 ms->handler = my_rules[i].handler;
1622 ms->state = MS_NEW;
1623 ms->fd = (-1);
1624 ms->url = NULL;
1625 ms->post = NULL;
1626 ms->conn = conn;
1627 *rctx = ms;
1628 return ms->handler(cctx,conn,url,method,version,
1629 data,data_size,rctx);
1630 }
1631
1632 if (!strcmp(method,"QUIT")) {
1633 (void)sem_post((sem_t *)cctx);
1634 return MHD_NO;
1635 }
1636
1637 fprintf(stderr,"bad request m=%s u=%s\n",method,url);
1638 free_ms(ms);
1639
1640 resp = MHD_create_response_from_data(0,NULL,MHD_NO,MHD_NO);
1641 if (!resp) {
1642 return MHD_NO;
1643 }
1644 MHD_queue_response(conn,MHD_HTTP_NOT_FOUND,resp);
1645 MHD_destroy_response(resp);
1646 return MHD_YES;
1647 }
1648
1649 /* These enum values cannot possibly conflict with the option values
1650 ordinarily used by commands, including CHAR_MAX + 1, etc. Avoid
1651 CHAR_MIN - 1, as it may equal -1, the getopt end-of-options value. */
1652 enum
1653 {
1654 GETOPT_HELP_CHAR = (CHAR_MIN - 2),
1655 GETOPT_VERSION_CHAR = (CHAR_MIN - 3)
1656 };
1657
1658 struct option my_options[] = {
1659 { "config", required_argument, NULL, 'c' },
1660 { "db", required_argument, NULL, 'd' },
1661 { "master", required_argument, NULL, 'm' },
1662 { "port", required_argument, NULL, 'p' },
1663 { "verbose", no_argument, NULL, 'v' },
1664 { "version", no_argument, NULL, GETOPT_VERSION_CHAR },
1665 { "help", no_argument, NULL, GETOPT_HELP_CHAR },
1666 { NULL, 0, NULL, '\0' }
1667 };
1668
1669 void
1670 usage (int status)
1671 {
1672 if (status != EXIT_SUCCESS)
1673 fprintf (stderr, _("Try `%s --help' for more information.\n"),
1674 program_name);
1675 else
1676 {
1677 printf (_("\
1678 Usage: %s [OPTION]\n\
1679 "),
1680 program_name);
1681 fputs (_("\
1682 Deltacloud image-warehouse daemon.\n\
1683 A configuration file must be specified.\n\
1684 \n\
1685 -c, --config=FILE config file [required]\n\
1686 -d, --db=HOST_PORT database server as ip[:port]\n\
1687 -m, --master=HOST_PORT master (upstream) server as ip[:port]\n\
1688 -p, --port=PORT alternate listen port (default 9090)\n\
1689 -v, --verbose verbose/debug output\n\
1690 \n\
1691 --help display this help and exit\n\
1692 --version output version information and exit\n\
1693 "), stdout);
1694 printf (_("\
1695 \n\
1696 Report %s bugs to %s.\n\
1697 "),
1698 program_name, PACKAGE_BUGREPORT);
1699 }
1700 exit (status);
1701 }
1702
1703 int
1704 main (int argc, char **argv)
1705 {
1706 struct MHD_Daemon *the_daemon;
1707 sem_t the_sem;
1708 char *stctx = NULL;
1709 char *port_tmp;
1710
1711 program_name = argv[0];
1712
1713 for (;;) switch (getopt_long(argc,argv,"c:d:m:p:v",my_options,NULL)) {
1714 case 'c':
1715 cfg_file = optarg;
1716 break;
1717 case 'd':
1718 assert (optarg);
1719 db_host = strtok_r(optarg,":",&stctx);
1720 port_tmp = strtok_r(NULL,":",&stctx);
1721 if (port_tmp) {
1722 db_port = (unsigned short)strtoul(port_tmp,NULL,10);
1723 }
1724 break;
1725 case 'm':
1726 assert (optarg);
1727 master_host = strtok_r(optarg,":",&stctx);
1728 port_tmp = strtok_r(NULL,":",&stctx);
1729 if (port_tmp) {
1730 master_port = (unsigned short)strtoul(port_tmp,NULL,10);
1731 }
1732 break;
1733 case 'p':
1734 my_port = (unsigned short)strtoul(optarg,NULL,10);
1735 break;
1736 case 'v':
1737 ++verbose;
1738 break;
1739 case GETOPT_HELP_CHAR:
1740 usage(EXIT_SUCCESS);
1741 break;
1742 case GETOPT_VERSION_CHAR:
1743 printf ("%s version %s\n", program_name, PACKAGE_VERSION);
1744 exit (EXIT_SUCCESS);
1745 break;
1746
1747 case -1:
1748 goto args_done;
1749 default:
1750 usage(EXIT_FAILURE);
1751 break;
1752 }
1753 args_done:
1754
1755 if (!cfg_file) {
1756 error (0, 0, "no configuration file specified");
1757 usage (EXIT_FAILURE);
1758 }
1759
1760 me = parse_config();
1761 if (!me) {
1762 fprintf(stderr,"could not parse %s\n",cfg_file);
1763 return !0;
1764 }
1765
1766 sem_init(&the_sem,0,0);
1767 if (proxy_host) {
1768 if (s3mode) {
1769 main_func_tbl = &s3_func_tbl;
1770 }
1771 else {
1772 main_func_tbl = &curl_func_tbl;
1773 }
1774 }
1775 else {
1776 main_func_tbl = &fs_func_tbl;
1777 }
1778
1779 if (verbose) {
1780 printf("primary store type is %s\n",main_func_tbl->name);
1781 if (master_host) {
1782 printf("operating as slave to %s:%u\n",
1783 master_host, master_port);
1784 }
1785 printf("db is at %s:%u\n",db_host,db_port);
1786 printf("will listen on port %u\n",my_port);
1787 printf("my location is \"%s\"\n",me);
1788 fflush(stdout);
1789 }
1790
1791 backend_init();
1792 main_func_tbl->init_func();
1793 meta_init();
1794 repl_init();
1795
1796 /*
1797 * Gotcha: if we don't set the connection memory limit explicitly,
1798 * the per-connection buffer for MHD will be smaller than that used
1799 * by CURL, so proxy_writefunc will never be able to do its job.
1800 */
1801 the_daemon = MHD_start_daemon(MY_MHD_FLAGS,
1802 my_port, NULL, NULL, &access_handler, &the_sem,
1803 MHD_OPTION_CONNECTION_MEMORY_LIMIT, (size_t)1048576,
1804 MHD_OPTION_END);
1805 if (!the_daemon) {
1806 fprintf(stderr,"Could not create daemon.\n");
1807 return !0;
1808 }
1809
1810 sem_wait(&the_sem);
1811 return 0;
1812 }
image warehouse daemon