?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
build: make the "rpm" rule work once again
[iwhd.git] / backend.c
blobc18d89548343367017dbf3f966375596c2ad7844
1 /* Copyright (C) 2010 Red Hat, 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 <fcntl.h>
19 #include <poll.h>
20 #include <pthread.h>
21 #include <regex.h>
22 #include <semaphore.h>
23 #include <stdint.h>
24 #include <stdio.h>
25 #include <stdlib.h>
26 #include <string.h>
27 #include <unistd.h>
28 #include <sys/stat.h>
29 #include <sys/wait.h>
30 #include <errno.h>
31 #include <error.h>
32
33 #include <microhttpd.h>
34 #include <curl/curl.h>
35 #include <hstor.h>
36 #include <glib.h>
37
38 #define GLOBALS_IMPL
39 #include "iwh.h"
40 #include "meta.h"
41 #include "setup.h"
42 #include "template.h"
43 #include "mpipe.h"
44 #include "backend.h"
45 #include "state_defs.h"
46
47 struct hstor_client *hstor;
48
49 /***** Generic module stuff, not specific to one back end *****/
50
51 /*
52 * Sizes for internal string buffers. In general, ADDR_SIZE needs to be
53 * big enough to hold a hostname, a port number, a bucket and key (each
54 * MAX_FIELD_LEN=64) and some punctuation. Header size needs to be big
55 * enough to hold the header name plus a CF token (32 bytes).
56 */
57 #define ADDR_SIZE 256
58 #define HEADER_SIZE 64
59
60 #define S3_IMAGE_PATTERN "^IMAGE[[:blank:]]+([^[:space:]]+)"
61 #define S3_ERROR_PATTERN "^ERROR[[:blank:]]+([^[:space:]]+)"
62
63 regex_t s3_success_pat;
64 regex_t s3_failure_pat;
65 int regex_ok = FALSE;
66
67 void
68 backend_init (void)
69 {
70 regex_ok = TRUE;
71
72 if (regcomp(&s3_success_pat,S3_IMAGE_PATTERN,REG_EXTENDED) != 0){
73 DPRINTF("could not compile S3 success pattern\n");
74 regex_ok = FALSE;
75 }
76
77 if (regcomp(&s3_failure_pat,S3_ERROR_PATTERN,REG_EXTENDED) != 0){
78 DPRINTF("could not compile S3 failure pattern\n");
79 regex_ok = FALSE;
80 }
81 }
82
83 /***** Stub functions for unimplemented stuff. *****/
84
85 static void
86 bad_init (provider_t *prov)
87 {
88 (void)prov;
89
90 DPRINTF("*** bad call to %s\n",__func__);
91 }
92
93 static void *
94 bad_get_child (void * ctx)
95 {
96 backend_thunk_t *tp = (backend_thunk_t *)ctx;
97 my_state *ms = tp->parent;
98
99 DPRINTF("*** bad call to %s\n",__func__);
100 pipe_prod_siginit(&ms->pipe,-1);
101 return NULL;
102 }
103
104 static void *
105 bad_put_child (void * ctx)
106 {
107 pipe_private *pp = ctx;
108 pipe_shared *ps = pp->shared;
109
110 DPRINTF("*** bad call to %s\n",__func__);
111 pipe_cons_siginit(ps, -1);
112 free(pp);
113 return THREAD_FAILED;
114 }
115
116 static void *
117 bad_cache_child (void * ctx)
118 {
119 (void)ctx;
120
121 DPRINTF("*** bad call to %s\n",__func__);
122 return NULL;
123 }
124
125 static int
126 bad_delete (const provider_t *prov, const char *bucket, const char *key,
127 const char *url)
128 {
129 (void)prov;
130 (void)bucket;
131 (void)key;
132 (void)url;
133
134 DPRINTF("*** bad call to %s\n",__func__);
135 return MHD_HTTP_BAD_REQUEST;
136 }
137
138 static int
139 bad_bcreate (const provider_t *prov, const char *bucket)
140 {
141 (void)prov;
142 (void)bucket;
143
144 DPRINTF("*** bad call to %s\n",__func__);
145 return MHD_HTTP_NOT_IMPLEMENTED;
146 }
147
148 static int
149 bad_register (my_state *ms, const provider_t *prov, const char *next,
150 GHashTable *args)
151 {
152 (void)ms;
153 (void)prov;
154 (void)next;
155 (void)args;
156
157 DPRINTF("*** bad call to %s\n",__func__);
158 return MHD_HTTP_NOT_IMPLEMENTED;
159 }
160
161 /***** Generic functions shared by the HTTP back ends. */
162
163 /* Invoked from S3/CURL/CF. */
164 static size_t
165 http_get_prod (void *ptr, size_t size, size_t nmemb, void *stream)
166 {
167 size_t total = size * nmemb;
168 pipe_shared *ps = stream;
169
170 DPRINTF("producer posting %zu bytes as %ld\n",total,ps->sequence+1);
171 pipe_prod_signal(ps,ptr,total);
172
173 DPRINTF("producer finished chunk\n");
174 return total;
175 }
176
177 /* Invoked from S3/CURL/CF. */
178 static size_t
179 http_put_cons (void *ptr, size_t size, size_t nmemb, void *stream)
180 {
181 size_t total = size * nmemb;
182 pipe_private *pp = stream;
183 pipe_shared *ps = pp->shared;
184 size_t done;
185
186 DPRINTF("consumer asked to read %zu\n",total);
187
188 if (!pipe_cons_wait(pp)) {
189 return 0;
190 }
191
192 DPRINTF("consumer offset %zu into %zu\n",
193 pp->offset, ps->data_len);
194 done = ps->data_len - pp->offset;
195 if (done > total) {
196 done = total;
197 }
198 memcpy(ptr,(char *)(ps->data_ptr)+pp->offset,done);
199 pp->offset += done;
200 DPRINTF("consumer copied %zu, new offset %zu\n",
201 done, pp->offset);
202 if (pp->offset == ps->data_len) {
203 DPRINTF("consumer finished chunk\n");
204 pipe_cons_signal(pp, 0);
205 }
206
207 return done;
208 }
209
210 /***** S3-specific functions *****/
211
212 static void
213 s3_init (provider_t *prov)
214 {
215 char svc_acc[128];
216 int chars;
217
218 chars = snprintf(svc_acc,sizeof(svc_acc),"%s:%u",prov->host,prov->port);
219 if (chars >= (int)sizeof(svc_acc)) {
220 error(0,0,"hostname %s too long in %s",prov->host,__func__);
221 return;
222 }
223 hstor = hstor_new(svc_acc,prov->host,prov->username,prov->password);
224 if (hstor) {
225 if (verbose) {
226 hstor->verbose = 1;
227 }
228 }
229 else {
230 DPRINTF("could not create S3 client\n");
231 }
232 }
233
234 /* Start an S3 _producer_. */
235 static void *
236 s3_get_child (void * ctx)
237 {
238 backend_thunk_t *tp = (backend_thunk_t *)ctx;
239 my_state *ms = tp->parent;
240
241 /* TBD: check existence before calling siginit */
242 pipe_prod_siginit(&ms->pipe,0);
243
244 hstor_get(hstor,ms->bucket,ms->key,http_get_prod,&ms->pipe,0);
245 /* TBD: check return value */
246
247 pipe_prod_finish(&ms->pipe);
248
249 DPRINTF("producer exiting\n");
250 return NULL;
251 }
252
253 /* Start an S3 _consumer_. */
254 static void *
255 s3_put_child (void * ctx)
256 {
257 pipe_private *pp = ctx;
258 pipe_shared *ps = pp->shared;
259 my_state *ms = ps->owner;
260 curl_off_t llen;
261 const char *clen;
262 bool rcb;
263
264 llen = (curl_off_t)MHD_SIZE_UNKNOWN;
265 if (ms->be_flags & BACKEND_GET_SIZE) {
266 clen = MHD_lookup_connection_value(
267 ms->conn, MHD_HEADER_KIND, "Content-Length");
268 if (clen) {
269 llen = strtoll(clen,NULL,10);
270 }
271 else {
272 error (0, 0, "missing Content-Length");
273 }
274 }
275
276 pipe_cons_siginit(ps, 0);
277 rcb = hstor_put(hstor,ms->bucket,ms->key,http_put_cons,llen,pp,NULL);
278 if (!rcb) {
279 DPRINTF("%s returning with error\n",__func__);
280 pipe_cons_siginit(ps, -1);
281 free(pp);
282 return THREAD_FAILED;
283 }
284
285 DPRINTF("%s returning\n",__func__);
286 free(pp);
287 return NULL;
288 }
289
290 static int
291 s3_delete (const provider_t *prov, const char *bucket, const char *key,
292 const char *url)
293 {
294 (void)prov;
295 (void)url;
296
297 hstor_del(hstor,bucket,key);
298 /* TBD: check return value */
299
300 return MHD_HTTP_OK;
301 }
302
303 static int
304 s3_bcreate (const provider_t *prov, const char *bucket)
305 {
306 (void)prov;
307
308 DPRINTF("creating bucket %s\n",bucket);
309
310 if (!hstor_add_bucket(hstor,bucket)) {
311 DPRINTF(" bucket create failed\n");
312 return MHD_HTTP_INTERNAL_SERVER_ERROR;
313 }
314
315 return MHD_HTTP_OK;
316 }
317
318 static const char *
319 s3_init_tmpfile (const char *value)
320 {
321 char *path;
322 int fd;
323 size_t len;
324 ssize_t written;
325
326 /* FIXME: do not hard-code /tmp. */
327 path = strdup("/tmp/iwtmp.XXXXXX");
328 if (!path) {
329 return NULL;
330 }
331
332 fd = mkstemp(path);
333 if (fd < 0) {
334 error (0, errno, "%s: failed to create file from template", path);
335 free(path);
336 return NULL;
337 }
338
339 len = strlen(value);
340 if (len > 0) {
341 written = write(fd,value,len);
342 close(fd);
343 if (written != (ssize_t)len) {
344 if (written < 0) {
345 error (0, errno, "failed to write to %s", path);
346 }
347 else {
348 error (0, errno,
349 "invalid write length %zd in %s",
350 written, __func__);
351 }
352 unlink(path);
353 free(path);
354 return NULL;
355 }
356 }
357
358 return path;
359 }
360
361 static int
362 s3_register (my_state *ms, const provider_t *prov, const char *next,
363 GHashTable *args)
364 {
365 char *kernel = g_hash_table_lookup(args,"kernel");
366 char *ramdisk = g_hash_table_lookup(args,"ramdisk");
367 char *api_key;
368 char *api_secret;
369 const char *ami_cert;
370 const char *ami_key;
371 const char *ami_uid;
372 const char *argv[12];
373 int argc = 0;
374 pid_t pid;
375 int organ[2];
376 FILE *fp;
377 char buf[ADDR_SIZE];
378 char *cval = NULL;
379 char *kval = NULL;
380 int rc = MHD_HTTP_BAD_REQUEST;
381 char *ami_bkt;
382 char ami_id_buf[64];
383 regmatch_t match[2];
384
385 if (!regex_ok) {
386 return MHD_HTTP_BAD_REQUEST;
387 }
388
389 if (next) {
390 DPRINTF("S3 register with next!=NULL\n");
391 goto cleanup;
392 }
393
394 DPRINTF("*** register %s/%s via %s (%s:%d)\n",
395 ms->bucket, ms->key, prov->name, prov->host, prov->port);
396 if (kernel) {
397 DPRINTF(" (using kernel %s)\n",kernel);
398 }
399 if (ramdisk) {
400 DPRINTF(" (using ramdisk %s)\n",ramdisk);
401 }
402
403 api_key = g_hash_table_lookup(args,"api-key");
404 if (!api_key) {
405 api_key = (char *)prov->username;
406 if (!api_key) {
407 error (0, 0, "missing EC2 API key");
408 goto cleanup;
409 }
410 }
411
412 api_secret = g_hash_table_lookup(args,"api-secret");
413 if (!api_secret) {
414 api_secret = (char *)prov->password;
415 if (!prov->password) {
416 error (0, 0, "missing EC2 API secret");
417 goto cleanup;
418 }
419 }
420
421 cval = g_hash_table_lookup(args,"ami-cert");
422 if (cval) {
423 ami_cert = s3_init_tmpfile(cval);
424 if (!ami_cert) {
425 goto cleanup;
426 }
427 }
428 else {
429 ami_cert = get_provider_value(prov,"ami-cert");
430 if (!ami_cert) {
431 error (0, 0, "missing EC2 AMI cert");
432 goto cleanup;
433 }
434 }
435
436 kval = g_hash_table_lookup(args,"ami-key");
437 if (kval) {
438 ami_key = s3_init_tmpfile(kval);
439 if (!ami_cert) {
440 goto cleanup;
441 }
442 }
443 else {
444 ami_key = get_provider_value(prov,"ami-key");
445 if (!ami_key) {
446 error (0, 0, "missing EC2 AMI key");
447 goto cleanup;
448 }
449 }
450
451 ami_uid = g_hash_table_lookup(args,"ami-uid");
452 if (!ami_uid) {
453 ami_uid = get_provider_value(prov,"ami-uid");
454 if (!ami_uid) {
455 error (0, 0, "missing EC2 AMI uid");
456 goto cleanup;
457 }
458 }
459
460 ami_bkt = g_hash_table_lookup(args,"ami-bkt");
461 if (!ami_bkt) {
462 ami_bkt = ms->bucket;
463 }
464
465 /*
466 * This is the point where we go from validation to execution. If we
467 * were double-forking so this could all be asynchronous, or for that
468 * matter to return an early 100-continue, this would probably be the
469 * place to do it. Even without that, we set the ami-id here so that
470 * the caller can know things are actually in progress.
471 */
472 sprintf(ami_id_buf,"pending %lld",(long long)time(NULL));
473 DPRINTF("temporary ami-id = \"%s\"\n",ami_id_buf);
474 (void)meta_set_value(ms->bucket,ms->key,"ami-id",ami_id_buf);
475 rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
476
477 const char *cmd = "dc-register-image";
478 argv[argc++] = cmd;
479 argv[argc++] = ms->bucket;
480 argv[argc++] = ms->key;
481 argv[argc++] = api_key;
482 argv[argc++] = api_secret;
483 argv[argc++] = ami_cert;
484 argv[argc++] = ami_key;
485 argv[argc++] = ami_uid;
486 argv[argc++] = ami_bkt;
487 argv[argc++] = kernel ? kernel : "_default_";
488 argv[argc++] = ramdisk ? ramdisk : "_default_";
489 argv[argc] = NULL;
490
491 DPRINTF("api-key = %s\n",api_key);
492 DPRINTF("api-secret = %s\n",api_secret);
493 DPRINTF("ami-cert = %s\n",ami_cert);
494 DPRINTF("ami-key = %s\n",ami_key);
495 DPRINTF("ami-uid = %s\n",ami_uid);
496 DPRINTF("ami-bkt = %s\n",ami_bkt);
497
498 if (pipe(organ) < 0) {
499 error (0, errno, "pipe creation failed");
500 goto cleanup;
501 }
502
503 pid = fork();
504 if (pid < 0) {
505 error (0, errno, "fork failed");
506 close(organ[0]);
507 close(organ[1]);
508 goto cleanup;
509 }
510
511 if (pid == 0) {
512 (void)dup2(organ[1],STDOUT_FILENO);
513 (void)dup2(organ[1],STDERR_FILENO);
514 execvp(cmd, (char* const*)argv);
515 error (EXIT_FAILURE, errno, "failed run command %s", cmd);
516 }
517
518 DPRINTF("waiting for child...\n");
519 if (waitpid(pid,NULL,0) < 0) {
520 error (0, errno, "waitpid failed");
521 }
522 /* TBD: check identity/status from waitpid */
523 DPRINTF("...child exited\n");
524
525 close(organ[1]);
526 fp = fdopen(organ[0],"r");
527 if (!fp) {
528 DPRINTF("could not open parent pipe\n");
529 close(organ[0]);
530 goto cleanup;
531 }
532 while (fgets(buf,sizeof(buf)-1,fp)) {
533 buf[sizeof(buf)-1] = '\0';
534 if (regexec(&s3_success_pat,buf,2,match,0) == 0) {
535 buf[match[1].rm_eo] = '\0';
536 DPRINTF("found AMI ID: %s\n",buf+match[1].rm_so);
537 sprintf(ami_id_buf,"OK %.60s",buf+match[1].rm_so);
538 rc = MHD_HTTP_OK;
539 }
540 else if (regexec(&s3_failure_pat,buf,2,match,0) == 0) {
541 buf[match[1].rm_eo] = '\0';
542 DPRINTF("found error marker: %s\n",buf+match[1].rm_so);
543 sprintf(ami_id_buf,"failed %.56s",buf+match[1].rm_so);
544 rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
545 }
546 else {
547 DPRINTF("ignoring line: <%s>\n",buf);
548 }
549 }
550 fclose(fp);
551
552 cleanup:
553 /*
554 * This is a bit tricky. If we found the cert in the HTTP request and
555 * succeeded in creating a temp file, then this condition will succeed.
556 * If we failed to create the temp file, or never found a cert
557 * anywhere, there will be no ami_cert to clean up. If we got a cert
558 * from the config, then ami_cert will be set but we'll (correctly)
559 * skip cleanup because cval is null.
560 */
561 if (cval && ami_cert) {
562 unlink(ami_cert);
563 free((char *)ami_cert);
564 }
565 /* Same reasoning as above, with kval/ami_key. */
566 if (kval && ami_key) {
567 unlink(ami_key);
568 free((char *)ami_key);
569 }
570 (void)meta_set_value(ms->bucket,ms->key,"ami-id",ami_id_buf);
571
572 return rc;
573 }
574
575 /***** CURL-specific functions *****/
576
577 static void
578 curl_init (provider_t *prov)
579 {
580 (void)prov;
581 }
582
583 /* Start a CURL _producer_. */
584 static void *
585 curl_get_child (void * ctx)
586 {
587 char fixed[ADDR_SIZE];
588 backend_thunk_t *tp = (backend_thunk_t *)ctx;
589 my_state *ms = tp->parent;
590 provider_t *prov = tp->prov;
591 CURL *curl;
592 int chars;
593
594 curl = curl_easy_init();
595 if (!curl) {
596 pipe_prod_siginit(&ms->pipe,-1);
597 return NULL; /* TBD: flag error somehow */
598 }
599 if (ms->from_master) {
600 chars = snprintf(fixed,sizeof(fixed),"http://%s:%u%s",
601 master_host, master_port, ms->url);
602 }
603 else {
604 chars = snprintf(fixed,sizeof(fixed),"http://%s:%u%s",
605 prov->host, prov->port, ms->url);
606 }
607 if (chars >= (int)sizeof(fixed)) {
608 error(0,0,"path too long in %s",__func__);
609 return NULL;
610 }
611 curl_easy_setopt(curl,CURLOPT_URL,fixed);
612 curl_easy_setopt(curl,CURLOPT_WRITEFUNCTION, http_get_prod);
613 curl_easy_setopt(curl,CURLOPT_WRITEDATA,&ms->pipe);
614 pipe_prod_siginit(&ms->pipe,0);
615
616 curl_easy_perform(curl);
617 curl_easy_getinfo(curl,CURLINFO_RESPONSE_CODE,&ms->rc);
618 pipe_prod_finish(&ms->pipe);
619
620 DPRINTF("producer exiting\n");
621 curl_easy_cleanup(curl);
622 return NULL;
623 }
624
625 /* Start a CURL _consumer_. */
626 static void *
627 curl_put_child (void * ctx)
628 {
629 pipe_private *pp = ctx;
630 pipe_shared *ps = pp->shared;
631 my_state *ms = ps->owner;
632 provider_t *prov = pp->prov;
633 curl_off_t llen;
634 char fixed[ADDR_SIZE];
635 CURL *curl;
636 const char *clen;
637 struct curl_slist *slist = NULL;
638 int chars;
639
640 llen = (curl_off_t)MHD_SIZE_UNKNOWN;
641 if (ms->be_flags & BACKEND_GET_SIZE) {
642 clen = MHD_lookup_connection_value(
643 ms->conn, MHD_HEADER_KIND, "Content-Length");
644 if (clen) {
645 llen = strtoll(clen,NULL,10);
646 }
647 else {
648 error (0, 0, "missing Content-Length");
649 }
650 }
651
652 /*
653 * This is how the iwhd at the other end knows this is a replication
654 * request and not just a PUT from some random user.
655 * TBD: add some auth* for this.
656 */
657 slist = curl_slist_append(slist,"X-redhat-role: master");
658
659 curl = curl_easy_init();
660 if (!curl) {
661 pipe_cons_siginit(ps, -1);
662 free(pp);
663 return THREAD_FAILED;
664 }
665 chars = snprintf(fixed,sizeof(fixed),
666 "http://%s:%u/%s/%s",prov->host,prov->port,ms->bucket,ms->key);
667 if (chars >= (int)sizeof(fixed)) {
668 error(0,0,"path too long in %s",__func__);
669 return NULL;
670 }
671 curl_easy_setopt(curl,CURLOPT_URL,fixed);
672 curl_easy_setopt(curl,CURLOPT_UPLOAD,1);
673 curl_easy_setopt(curl,CURLOPT_INFILESIZE_LARGE,llen);
674 curl_easy_setopt(curl,CURLOPT_READFUNCTION,http_put_cons);
675 curl_easy_setopt(curl,CURLOPT_READDATA,pp);
676 curl_easy_setopt(curl,CURLOPT_HTTPHEADER,slist);
677 pipe_cons_siginit(ps, 0);
678 curl_easy_perform(curl);
679 curl_easy_cleanup(curl);
680 curl_slist_free_all(slist);
681
682 DPRINTF("%s returning\n",__func__);
683 free(pp);
684 return NULL;
685 }
686
687 /* Start a CURL cache consumer. */
688 static void *
689 curl_cache_child (void * ctx)
690 {
691 pipe_private *pp = ctx;
692 pipe_shared *ps = pp->shared;
693 my_state *ms = ps->owner;
694 provider_t *prov = pp->prov;
695 char fixed[ADDR_SIZE];
696 CURL *curl;
697 char *slash;
698 char *my_url = strdup(ms->url);
699 int chars;
700
701 if (!my_url) {
702 return THREAD_FAILED;
703 }
704
705 curl = curl_easy_init();
706 if (!curl) {
707 free(my_url);
708 pipe_cons_siginit(ps,-1);
709 return THREAD_FAILED;
710 }
711 chars = snprintf(fixed,sizeof(fixed),
712 "http://%s:%u%s",prov->host,prov->port,ms->url);
713 if (chars >= (int)sizeof(fixed)) {
714 error(0,0,"path too long in %s",__func__);
715 return NULL;
716 }
717 curl_easy_setopt(curl,CURLOPT_URL,fixed);
718 curl_easy_setopt(curl,CURLOPT_UPLOAD,1);
719 curl_easy_setopt(curl,CURLOPT_INFILESIZE_LARGE,
720 (curl_off_t)MHD_SIZE_UNKNOWN);
721 curl_easy_setopt(curl,CURLOPT_READFUNCTION,http_put_cons);
722 curl_easy_setopt(curl,CURLOPT_READDATA,pp);
723 curl_easy_perform(curl);
724 curl_easy_cleanup(curl);
725
726 slash = strchr(my_url+1,'/');
727 if (slash) {
728 *slash = '\0';
729 meta_got_copy(my_url+1,slash+1,me);
730 }
731
732 free(my_url);
733 return NULL;
734 }
735
736 static int
737 curl_delete (const provider_t *prov, const char *bucket, const char *key,
738 const char *url)
739 {
740 CURL *curl;
741 char fixed[ADDR_SIZE];
742 int chars;
743
744 (void)bucket;
745 (void)key;
746
747 curl = curl_easy_init();
748 if (!curl) {
749 return MHD_HTTP_INTERNAL_SERVER_ERROR;
750 }
751
752 chars = snprintf(fixed,sizeof(fixed),
753 "http://%s:%u%s",prov->host,prov->port,url);
754 if (chars >= (int)sizeof(fixed)) {
755 error(0,0,"path too long in %s",__func__);
756 return MHD_HTTP_INTERNAL_SERVER_ERROR;
757 }
758 curl_easy_setopt(curl,CURLOPT_URL,fixed);
759 curl_easy_setopt(curl,CURLOPT_CUSTOMREQUEST,"DELETE");
760 curl_easy_perform(curl);
761 curl_easy_cleanup(curl);
762
763 return MHD_HTTP_OK;
764 }
765
766 static int
767 curl_bcreate (const provider_t *prov, const char *bucket)
768 {
769 char addr[ADDR_SIZE];
770 int chars;
771 CURL *curl;
772
773 chars = snprintf(addr,sizeof(addr),"http://%s:%d/%s",
774 prov->host,prov->port,bucket);
775 if (chars >= (int)sizeof(addr)) {
776 error(0,0,"path too long in %s",__func__);
777 return MHD_HTTP_INTERNAL_SERVER_ERROR;
778 }
779
780 curl = curl_easy_init();
781 if (!curl) {
782 error(0,errno,"no memory in %s",__func__);
783 return MHD_HTTP_INTERNAL_SERVER_ERROR;
784 }
785 curl_easy_setopt(curl,CURLOPT_URL,addr);
786 curl_easy_setopt(curl,CURLOPT_CUSTOMREQUEST,"PUT");
787 curl_easy_perform(curl);
788 curl_easy_cleanup(curl);
789 return MHD_HTTP_OK;
790 }
791
792 /*
793 * We can proxy through any number of CURL/HTTP warehouses, but the chain
794 * eventually has to terminate at an S3 back end.
795 */
796
797 static int
798 curl_register (my_state *ms, const provider_t *prov, const char *next,
799 GHashTable *args)
800 {
801 char fixed[ADDR_SIZE];
802 CURL *curl;
803 struct curl_httppost *first = NULL;
804 struct curl_httppost *last = NULL;
805 char *kernel = g_hash_table_lookup(args,"kernel");
806 char *ramdisk = g_hash_table_lookup(args,"ramdisk");
807 int chars;
808
809 if (!next) {
810 DPRINTF("CURL register with next==NULL\n");
811 return MHD_HTTP_BAD_REQUEST;
812 }
813
814 DPRINTF("*** PROXY registration request for %s/%s to %s (%s:%d)\n",
815 ms->bucket, ms->key, prov->name, prov->host, prov->port);
816
817 curl = curl_easy_init();
818 if (!curl) {
819 return MHD_HTTP_INTERNAL_SERVER_ERROR;
820 }
821 chars = snprintf(fixed,sizeof(fixed),"http://%s:%d/%s/%s",
822 prov->host,prov->port, ms->bucket, ms->key);
823 if (chars >= (int)sizeof(fixed)) {
824 error(0,0,"path too long in %s",__func__);
825 return MHD_HTTP_INTERNAL_SERVER_ERROR;
826 }
827 curl_easy_setopt(curl,CURLOPT_URL,fixed);
828 curl_formadd(&first,&last,
829 CURLFORM_COPYNAME, "op",
830 CURLFORM_COPYCONTENTS, "register",
831 CURLFORM_END);
832 curl_formadd(&first,&last,
833 CURLFORM_COPYNAME, "site",
834 CURLFORM_COPYCONTENTS, next,
835 CURLFORM_END);
836 if (kernel) {
837 curl_formadd(&first,&last,
838 CURLFORM_COPYNAME, "kernel",
839 CURLFORM_COPYCONTENTS, kernel,
840 CURLFORM_END);
841 }
842 if (ramdisk) {
843 curl_formadd(&first,&last,
844 CURLFORM_COPYNAME, "ramdisk",
845 CURLFORM_COPYCONTENTS, ramdisk,
846 CURLFORM_END);
847 }
848 curl_easy_setopt(curl,CURLOPT_HTTPPOST,first);
849 curl_easy_perform(curl);
850 curl_easy_cleanup(curl);
851
852 return MHD_HTTP_OK;
853 }
854
855 /***** CF-specific functions *****/
856
857 /*
858 * TBD: refactor to maximize common code. Despite the de-duplication between
859 * this module and replica.c, there's still a lot more that could be done to
860 * combine xxx_yyy_child for xxx={http,cf} and yyy={put,cache}. A rough
861 * outline might be:
862 *
863 * if xxx=cf, call CF-specific routine to add CF auth header
864 * do common curl setup and execution
865 * if yyy=cache, call meta_got_copy
866 *
867 * There might even be an opportunity to combine code for put and bucket
868 * create in some cases, since the only difference is the URL and the
869 * lack of a data transfer in the bucket-create case.
870 */
871
872 static size_t
873 cf_writer (void *ptr ATTRIBUTE_UNUSED, size_t size, size_t nmemb,
874 void *stream ATTRIBUTE_UNUSED)
875 {
876 return size * nmemb;
877 }
878
879 static size_t
880 cf_header (void *ptr, size_t size, size_t nmemb, void *stream)
881 {
882 char *next;
883 char *sctx;
884 provider_t *prov = (provider_t *)stream;
885
886 next = strtok_r(ptr,":",&sctx);
887 if (next) {
888 if (!strcasecmp(next,"X-Storage-Url")) {
889 next = strtok_r(NULL," \n\r",&sctx);
890 if (next) {
891 DPRINTF("got CF URL %s\n",next);
892 /* NB: after this, original "host" is gone. */
893 free((char *)prov->host);
894 prov->host = strdup(next);
895 }
896 }
897 else if (!strcasecmp(next,"X-Storage-Token")) {
898 next = strtok_r(NULL," \n\r",&sctx);
899 if (next) {
900 DPRINTF("got CF token %s\n",next);
901 prov->token = strdup(next);
902 }
903 }
904 }
905 return size * nmemb;
906 }
907
908 static struct curl_slist *
909 cf_add_token (struct curl_slist *in_slist, const char *token)
910 {
911 int chars;
912 char auth_hdr[HEADER_SIZE];
913
914 if (!token) {
915 return in_slist;
916 }
917
918 chars = snprintf(auth_hdr,sizeof(auth_hdr),"X-Auth-Token: %s",token);
919 if (chars >= (int)sizeof(auth_hdr)) {
920 error(0,0,"auth_hdr too long");
921 return in_slist;
922 }
923
924 return curl_slist_append(NULL,auth_hdr);
925 }
926
927 static void
928 cf_init (provider_t *prov)
929 {
930 CURL *curl;
931 char addr[ADDR_SIZE];
932 char auth_user[HEADER_SIZE];
933 char auth_key[HEADER_SIZE];
934 struct curl_slist *slist;
935 int chars;
936
937 if (prov->token) {
938 return;
939 }
940
941 chars = snprintf(addr,sizeof(addr),"https://%s:%u/v1.0",
942 prov->host, prov->port);
943 if (chars >= (int)sizeof(addr)) {
944 error(0,0,"API URL too long in %s",__func__);
945 return;
946 }
947
948 chars = snprintf(auth_user,sizeof(auth_user),"X-Auth-User: %s",
949 prov->username);
950 if (chars >= (int)sizeof(auth_user)) {
951 error(0,0,"auth_user too long in %s",__func__);
952 return;
953 }
954
955 chars = snprintf(auth_key,sizeof(auth_key),"X-Auth-Key: %s",
956 prov->password);
957 if (chars >= (int)sizeof(auth_key)) {
958 error(0,0,"auth_key too long in %s",__func__);
959 return;
960 }
961
962 curl = curl_easy_init();
963 curl_easy_setopt(curl,CURLOPT_URL,addr);
964 curl_easy_setopt(curl,CURLOPT_WRITEFUNCTION,cf_writer);
965 curl_easy_setopt(curl,CURLOPT_HEADERFUNCTION,cf_header);
966 curl_easy_setopt(curl,CURLOPT_WRITEHEADER,prov);
967 slist = curl_slist_append(NULL,auth_user);
968 slist = curl_slist_append(slist,auth_key);
969 curl_easy_setopt(curl,CURLOPT_HTTPHEADER,slist);
970 curl_easy_perform(curl);
971 curl_easy_cleanup(curl);
972 curl_slist_free_all(slist);
973
974 DPRINTF("CF token = %s\n",prov->token);
975 }
976
977 /* Start a CloudFiles _producer_. */
978 static void *
979 cf_get_child (void * ctx)
980 {
981 char fixed[ADDR_SIZE];
982 backend_thunk_t *tp = (backend_thunk_t *)ctx;
983 my_state *ms = tp->parent;
984 provider_t *prov = tp->prov;
985 CURL *curl;
986 struct curl_slist *slist = NULL;
987 int chars;
988
989 slist = cf_add_token(slist,prov->token);
990 if (!slist) {
991 return THREAD_FAILED;
992 }
993 /*
994 * Rackspace doesn't clearly document that you'll get
995 * 412 (Precondition Failed) if you omit this.
996 */
997 slist = curl_slist_append(slist,
998 "Content-Type: binary/octet-stream");
999
1000 curl = curl_easy_init();
1001 if (!curl) {
1002 pipe_prod_siginit(&ms->pipe,-1);
1003 curl_slist_free_all(slist);
1004 return NULL; /* TBD: flag error somehow */
1005 }
1006 chars = snprintf(fixed,sizeof(fixed),"%s%s", prov->host, ms->url);
1007 if (chars >= (int)sizeof(fixed)) {
1008 error(0,0,"path too long in %s",__func__);
1009 return NULL;
1010 }
1011 curl_easy_setopt(curl,CURLOPT_URL,fixed);
1012 curl_easy_setopt(curl,CURLOPT_WRITEFUNCTION, http_get_prod);
1013 curl_easy_setopt(curl,CURLOPT_WRITEDATA,&ms->pipe);
1014 curl_easy_setopt(curl,CURLOPT_HTTPHEADER,slist);
1015 pipe_prod_siginit(&ms->pipe,0);
1016
1017 curl_easy_perform(curl);
1018 curl_easy_getinfo(curl,CURLINFO_RESPONSE_CODE,&ms->rc);
1019 pipe_prod_finish(&ms->pipe);
1020
1021 DPRINTF("producer exiting\n");
1022 curl_easy_cleanup(curl);
1023 curl_slist_free_all(slist);
1024 return NULL;
1025 }
1026
1027 /* Start a CloudFiles _consumer_. */
1028 static void *
1029 cf_put_child (void * ctx)
1030 {
1031 pipe_private *pp = ctx;
1032 pipe_shared *ps = pp->shared;
1033 my_state *ms = ps->owner;
1034 provider_t *prov = pp->prov;
1035 curl_off_t llen;
1036 char fixed[ADDR_SIZE];
1037 CURL *curl;
1038 const char *clen;
1039 struct curl_slist *slist = NULL;
1040 int chars;
1041
1042 slist = cf_add_token(slist,prov->token);
1043 if (!slist) {
1044 return THREAD_FAILED;
1045 }
1046
1047 llen = (curl_off_t)MHD_SIZE_UNKNOWN;
1048 if (ms->be_flags & BACKEND_GET_SIZE) {
1049 clen = MHD_lookup_connection_value(
1050 ms->conn, MHD_HEADER_KIND, "Content-Length");
1051 if (clen) {
1052 llen = strtoll(clen,NULL,10);
1053 }
1054 else {
1055 error (0, 0, "missing Content-Length");
1056 }
1057 }
1058
1059 curl = curl_easy_init();
1060 if (!curl) {
1061 pipe_cons_siginit(ps, -1);
1062 free(pp);
1063 curl_slist_free_all(slist);
1064 return THREAD_FAILED;
1065 }
1066 chars = snprintf(fixed,sizeof(fixed),
1067 "%s/%s/%s",prov->host,ms->bucket,ms->key);
1068 if (chars >= (int)sizeof(fixed)) {
1069 error(0,0,"path too long in %s",__func__);
1070 return NULL;
1071 }
1072 curl_easy_setopt(curl,CURLOPT_URL,fixed);
1073 curl_easy_setopt(curl,CURLOPT_UPLOAD,1);
1074 curl_easy_setopt(curl,CURLOPT_INFILESIZE_LARGE,llen);
1075 curl_easy_setopt(curl,CURLOPT_READFUNCTION,http_put_cons);
1076 curl_easy_setopt(curl,CURLOPT_READDATA,pp);
1077 curl_easy_setopt(curl,CURLOPT_HTTPHEADER,slist);
1078 pipe_cons_siginit(ps, 0);
1079
1080 curl_easy_perform(curl);
1081 curl_easy_getinfo(curl,CURLINFO_RESPONSE_CODE,&ms->rc);
1082
1083 DPRINTF("%s returning\n",__func__);
1084 curl_easy_cleanup(curl);
1085 curl_slist_free_all(slist);
1086 free(pp);
1087 return NULL;
1088 }
1089
1090 static int
1091 cf_delete (const provider_t *prov,
1092 const char *bucket ATTRIBUTE_UNUSED,
1093 const char *key ATTRIBUTE_UNUSED,
1094 const char *url)
1095 {
1096 CURL *curl;
1097 char fixed[ADDR_SIZE];
1098 long rc;
1099 struct curl_slist *slist = NULL;
1100 int chars;
1101
1102 slist = cf_add_token(slist,prov->token);
1103 if (!slist) {
1104 return MHD_HTTP_INTERNAL_SERVER_ERROR;
1105 }
1106
1107 curl = curl_easy_init();
1108 if (!curl) {
1109 curl_slist_free_all(slist);
1110 return MHD_HTTP_INTERNAL_SERVER_ERROR;
1111 }
1112
1113 chars = snprintf(fixed,sizeof(fixed),"%s%s",prov->host,url);
1114 if (chars >= (int)sizeof(fixed)) {
1115 error(0,0,"path too long in %s",__func__);
1116 return MHD_HTTP_INTERNAL_SERVER_ERROR;
1117 }
1118 curl_easy_setopt(curl,CURLOPT_URL,fixed);
1119 curl_easy_setopt(curl,CURLOPT_CUSTOMREQUEST,"DELETE");
1120 curl_easy_setopt(curl,CURLOPT_HTTPHEADER,slist);
1121
1122 curl_easy_perform(curl);
1123 curl_easy_getinfo(curl,CURLINFO_RESPONSE_CODE,&rc);
1124 DPRINTF("%s: rc = %ld\n",__func__,rc);
1125
1126 curl_easy_cleanup(curl);
1127 curl_slist_free_all(slist);
1128
1129 return MHD_HTTP_OK;
1130 }
1131
1132 static size_t
1133 cf_null_reader (void *ptr ATTRIBUTE_UNUSED,
1134 size_t size ATTRIBUTE_UNUSED,
1135 size_t nmemb ATTRIBUTE_UNUSED,
1136 void *stream ATTRIBUTE_UNUSED)
1137 {
1138 return 0;
1139 }
1140
1141 static int
1142 cf_bcreate (const provider_t *prov, const char *bucket)
1143 {
1144 char fixed[ADDR_SIZE];
1145 CURL *curl;
1146 long rc;
1147 struct curl_slist *slist = NULL;
1148 int chars;
1149
1150 slist = cf_add_token(slist,prov->token);
1151 if (!slist) {
1152 return MHD_HTTP_INTERNAL_SERVER_ERROR;
1153 }
1154
1155 curl = curl_easy_init();
1156 if (!curl) {
1157 curl_slist_free_all(slist);
1158 return MHD_HTTP_INTERNAL_SERVER_ERROR;
1159 }
1160 chars = snprintf(fixed,sizeof(fixed),"%s/%s",prov->host,bucket);
1161 if (chars >= (int)sizeof(fixed)) {
1162 error(0,0,"path too long in %s",__func__);
1163 return MHD_HTTP_INTERNAL_SERVER_ERROR;
1164 }
1165 curl_easy_setopt(curl,CURLOPT_URL,fixed);
1166 curl_easy_setopt(curl,CURLOPT_UPLOAD,1);
1167 curl_easy_setopt(curl,CURLOPT_INFILESIZE_LARGE,
1168 (curl_off_t)MHD_SIZE_UNKNOWN);
1169 curl_easy_setopt(curl,CURLOPT_READFUNCTION,cf_null_reader);
1170 curl_easy_setopt(curl,CURLOPT_HTTPHEADER,slist);
1171
1172 curl_easy_perform(curl);
1173 curl_easy_getinfo(curl,CURLINFO_RESPONSE_CODE,&rc);
1174 DPRINTF("%s: rc = %ld\n",__func__,rc);
1175
1176 DPRINTF("%s returning\n",__func__);
1177 curl_easy_cleanup(curl);
1178 curl_slist_free_all(slist);
1179 return MHD_HTTP_OK;
1180 }
1181
1182 /***** FS-specific functions *****/
1183
1184 static void
1185 fs_init (provider_t *prov)
1186 {
1187 DPRINTF("changing directory to %s\n",prov->path);
1188 if (chdir(prov->path) < 0) {
1189 error(0,errno,"chdir failed, unsafe to continue");
1190 exit(!0); /* Value doesn't matter, as long as it's not zero. */
1191 }
1192 }
1193
1194 /* Start an FS _producer_. */
1195 static void *
1196 fs_get_child (void * ctx)
1197 {
1198 backend_thunk_t *tp = (backend_thunk_t *)ctx;
1199 my_state *ms = tp->parent;
1200 int fd;
1201 char buf[1<<16];
1202 ssize_t bytes;
1203 char *file = ms->url+1;
1204
1205 fd = open(file, O_RDONLY);
1206 if (fd < 0) {
1207 pipe_prod_siginit(&ms->pipe,-1);
1208 pipe_prod_finish(&ms->pipe);
1209 return THREAD_FAILED;
1210 }
1211
1212 pipe_prod_siginit(&ms->pipe,0);
1213
1214 for (;;) {
1215 bytes = read(fd,buf,sizeof(buf));
1216 if (bytes <= 0) {
1217 if (bytes < 0) {
1218 error (0, errno, "%s: read failed", file);
1219 }
1220 break;
1221 }
1222 pipe_prod_signal(&ms->pipe,buf,bytes);
1223 }
1224
1225 close(fd);
1226 pipe_prod_finish(&ms->pipe);
1227
1228 DPRINTF("producer exiting\n");
1229 return NULL;
1230 }
1231
1232 /* Start an FS _consumer_. */
1233 static void *
1234 fs_put_child (void * ctx)
1235 {
1236 pipe_private *pp = ctx;
1237 pipe_shared *ps = pp->shared;
1238 my_state *ms = ps->owner;
1239 int fd;
1240 ssize_t bytes;
1241 size_t offset;
1242 char fixed[ADDR_SIZE];
1243 int chars;
1244
1245 chars = snprintf(fixed,sizeof(fixed),"%s/%s",ms->bucket,ms->key);
1246 if (chars >= (int)sizeof(fixed)) {
1247 error(0,0,"path too long in %s",__func__);
1248 return NULL;
1249 }
1250 if (unlink(fixed) < 0) {
1251 error(0,errno,"unlink failed for %s (non-fatal)",fixed);
1252 }
1253 fd = open(fixed,O_WRONLY|O_CREAT|O_EXCL,0666);
1254 if (fd < 0) {
1255 pipe_cons_siginit(ps, errno);
1256 free(pp);
1257 return THREAD_FAILED;
1258 }
1259
1260 pipe_cons_siginit(ps, 0);
1261
1262 while (pipe_cons_wait(pp)) {
1263 for (offset = 0; offset < ps->data_len; offset += bytes) {
1264 bytes = write(fd,
1265 (char *)(ps->data_ptr)+offset,
1266 ps->data_len-offset);
1267 if (bytes <= 0) {
1268 if (bytes < 0) {
1269 error (0, errno, "%s: write failed",
1270 fixed);
1271 pipe_cons_signal(pp, errno);
1272 }
1273 else {
1274 pipe_cons_signal(pp, ENOSPC);
1275 }
1276 break;
1277 }
1278 }
1279 pipe_cons_signal(pp, 0);
1280 }
1281
1282 close(fd);
1283
1284 DPRINTF("%s returning\n",__func__);
1285 free(pp);
1286 return NULL;
1287 }
1288
1289 static int
1290 fs_delete (const provider_t *prov, const char *bucket, const char *key,
1291 const char *url)
1292 {
1293 (void)prov;
1294 (void)bucket;
1295 (void)key;
1296
1297 if (unlink(url+1) < 0) {
1298 error (0, errno, "%s: failed to unlink", url+1);
1299 return MHD_HTTP_NOT_FOUND;
1300 }
1301
1302 return MHD_HTTP_OK;
1303 }
1304
1305 static int
1306 fs_bcreate (const provider_t *prov, const char *bucket)
1307 {
1308 (void)prov;
1309
1310 DPRINTF("creating bucket %s\n",bucket);
1311
1312 if (mkdir(bucket,0700) < 0) {
1313 error (0, errno, "%s: failed to create directory", bucket);
1314 return MHD_HTTP_INTERNAL_SERVER_ERROR;
1315 }
1316
1317 return MHD_HTTP_OK;
1318 }
1319
1320 /***** Function tables. ****/
1321
1322 backend_func_tbl bad_func_tbl = {
1323 "uninitialized",
1324 bad_init,
1325 bad_get_child,
1326 bad_put_child,
1327 bad_cache_child,
1328 bad_delete,
1329 bad_bcreate,
1330 bad_register,
1331 };
1332
1333 backend_func_tbl s3_func_tbl = {
1334 "S3",
1335 s3_init,
1336 s3_get_child,
1337 s3_put_child,
1338 bad_cache_child,
1339 s3_delete,
1340 s3_bcreate,
1341 s3_register,
1342 };
1343
1344 backend_func_tbl curl_func_tbl = {
1345 "HTTP",
1346 curl_init,
1347 curl_get_child,
1348 curl_put_child,
1349 curl_cache_child,
1350 curl_delete,
1351 curl_bcreate,
1352 curl_register,
1353 };
1354
1355 backend_func_tbl cf_func_tbl = {
1356 "CF",
1357 cf_init,
1358 cf_get_child,
1359 cf_put_child,
1360 bad_cache_child,
1361 cf_delete,
1362 cf_bcreate,
1363 bad_register,
1364 };
1365
1366 backend_func_tbl fs_func_tbl = {
1367 "FS",
1368 fs_init,
1369 fs_get_child,
1370 fs_put_child,
1371 bad_cache_child,
1372 fs_delete,
1373 fs_bcreate,
1374 bad_register,
1375 };
image warehouse daemon