1 /* Copyright (C) 2010 Red Hat, Inc.
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.
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.
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/>. */
23 #include <semaphore.h>
36 #include <microhttpd.h>
37 #include <curl/curl.h>
48 #include "state_defs.h"
50 struct hstor_client
*hstor
;
52 /***** Generic module stuff, not specific to one back end *****/
54 /* Sizes for internal string buffers for CF code. */
55 #define ADDR_SIZE 1024
56 #define SVC_ACC_SIZE 128
57 #define HEADER_SIZE 64
59 #define S3_IMAGE_PATTERN "^IMAGE[[:blank:]]+([^[:space:]]+)"
60 #define S3_ERROR_PATTERN "^ERROR[[:blank:]]+([^[:space:]]+)"
62 regex_t s3_success_pat
;
63 regex_t s3_failure_pat
;
71 if (regcomp(&s3_success_pat
,S3_IMAGE_PATTERN
,REG_EXTENDED
) != 0){
72 DPRINTF("could not compile S3 success pattern\n");
76 if (regcomp(&s3_failure_pat
,S3_ERROR_PATTERN
,REG_EXTENDED
) != 0){
77 DPRINTF("could not compile S3 failure pattern\n");
83 /***** Stub functions for unimplemented stuff. *****/
86 bad_init (provider_t
*prov
)
90 DPRINTF("*** bad call to %s\n",__func__
);
94 bad_get_child (void * ctx
)
96 backend_thunk_t
*tp
= (backend_thunk_t
*)ctx
;
97 my_state
*ms
= tp
->parent
;
99 DPRINTF("*** bad call to %s\n",__func__
);
100 pipe_prod_siginit(&ms
->pipe
,-1);
105 bad_put_child (void * ctx
)
107 pipe_private
*pp
= ctx
;
108 pipe_shared
*ps
= pp
->shared
;
110 DPRINTF("*** bad call to %s\n",__func__
);
111 pipe_cons_siginit(ps
, -1);
113 return THREAD_FAILED
;
117 bad_cache_child (void * ctx
)
121 DPRINTF("*** bad call to %s\n",__func__
);
126 bad_delete (const provider_t
*prov
, const char *bucket
, const char *key
,
134 DPRINTF("*** bad call to %s\n",__func__
);
135 return MHD_HTTP_BAD_REQUEST
;
139 bad_bcreate (const provider_t
*prov
, const char *bucket
)
144 DPRINTF("*** bad call to %s\n",__func__
);
145 return MHD_HTTP_NOT_IMPLEMENTED
;
149 bad_register (my_state
*ms
, const provider_t
*prov
, const char *next
,
157 DPRINTF("*** bad call to %s\n",__func__
);
158 return MHD_HTTP_NOT_IMPLEMENTED
;
161 /***** Generic functions shared by the HTTP back ends. */
163 /* Invoked from S3/CURL/CF. */
165 http_get_prod (void *ptr
, size_t size
, size_t nmemb
, void *stream
)
167 size_t total
= size
* nmemb
;
168 pipe_shared
*ps
= stream
;
170 DPRINTF("producer posting %zu bytes as %ld\n",total
,ps
->sequence
+1);
171 pipe_prod_signal(ps
,ptr
,total
);
173 DPRINTF("producer finished chunk\n");
177 /* Invoked from S3/CURL/CF. */
179 http_put_cons (void *ptr
, size_t size
, size_t nmemb
, void *stream
)
181 size_t total
= size
* nmemb
;
182 pipe_private
*pp
= stream
;
183 pipe_shared
*ps
= pp
->shared
;
186 DPRINTF("consumer asked to read %zu\n",total
);
188 if (!pipe_cons_wait(pp
)) {
192 DPRINTF("consumer offset %zu into %zu\n",
193 pp
->offset
, ps
->data_len
);
194 done
= ps
->data_len
- pp
->offset
;
198 memcpy(ptr
,ps
->data_ptr
+pp
->offset
,done
);
200 DPRINTF("consumer copied %zu, new offset %zu\n",
202 if (pp
->offset
== ps
->data_len
) {
203 DPRINTF("consumer finished chunk\n");
204 pipe_cons_signal(pp
, 0);
210 /***** S3-specific functions *****/
213 s3_init (provider_t
*prov
)
217 snprintf(svc_acc
,sizeof(svc_acc
),"%s:%u",prov
->host
,prov
->port
);
218 hstor
= hstor_new(svc_acc
,prov
->host
,prov
->username
,prov
->password
);
225 DPRINTF("could not create S3 client\n");
229 /* Start an S3 _producer_. */
231 s3_get_child (void * ctx
)
233 backend_thunk_t
*tp
= (backend_thunk_t
*)ctx
;
234 my_state
*ms
= tp
->parent
;
236 /* TBD: check existence before calling siginit */
237 pipe_prod_siginit(&ms
->pipe
,0);
239 hstor_get(hstor
,ms
->bucket
,ms
->key
,http_get_prod
,&ms
->pipe
,0);
240 /* TBD: check return value */
242 pipe_prod_finish(&ms
->pipe
);
244 DPRINTF("producer exiting\n");
248 /* Start an S3 _consumer_. */
250 s3_put_child (void * ctx
)
252 pipe_private
*pp
= ctx
;
253 pipe_shared
*ps
= pp
->shared
;
254 my_state
*ms
= ps
->owner
;
259 llen
= (curl_off_t
)MHD_SIZE_UNKNOWN
;
260 if (ms
->be_flags
& BACKEND_GET_SIZE
) {
261 clen
= MHD_lookup_connection_value(
262 ms
->conn
, MHD_HEADER_KIND
, "Content-Length");
264 llen
= strtoll(clen
,NULL
,10);
267 error (0, 0, "missing Content-Length");
271 pipe_cons_siginit(ps
, 0);
272 rcb
= hstor_put(hstor
,ms
->bucket
,ms
->key
,http_put_cons
,llen
,pp
,NULL
);
274 DPRINTF("%s returning with error\n",__func__
);
275 pipe_cons_siginit(ps
, -1);
277 return THREAD_FAILED
;
280 DPRINTF("%s returning\n",__func__
);
286 s3_delete (const provider_t
*prov
, const char *bucket
, const char *key
,
292 hstor_del(hstor
,bucket
,key
);
293 /* TBD: check return value */
299 s3_bcreate (const provider_t
*prov
, const char *bucket
)
303 DPRINTF("creating bucket %s\n",bucket
);
305 if (!hstor_add_bucket(hstor
,bucket
)) {
306 DPRINTF(" bucket create failed\n");
307 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
314 s3_init_tmpfile (char *value
)
321 path
= strdup("/tmp/iwtmp.XXXXXX");
328 error (0, errno
, "%s: failed to create file from template", path
);
335 written
= write(fd
,value
,len
);
337 if (written
!= (ssize_t
)len
) {
339 error (0, errno
, "failed to write to %s", path
);
343 "invalid write length %zd in %s",
356 s3_register (my_state
*ms
, const provider_t
*prov
, const char *next
,
359 char *kernel
= g_hash_table_lookup(args
,"kernel");
360 char *ramdisk
= g_hash_table_lookup(args
,"ramdisk");
363 const char *ami_cert
;
366 const char *argv
[12];
374 int rc
= MHD_HTTP_BAD_REQUEST
;
380 return MHD_HTTP_BAD_REQUEST
;
384 DPRINTF("S3 register with next!=NULL\n");
388 DPRINTF("*** register %s/%s via %s (%s:%d)\n",
389 ms
->bucket
, ms
->key
, prov
->name
, prov
->host
, prov
->port
);
391 DPRINTF(" (using kernel %s)\n",kernel
);
394 DPRINTF(" (using ramdisk %s)\n",ramdisk
);
397 api_key
= g_hash_table_lookup(args
,"api-key");
399 api_key
= (char *)prov
->username
;
401 error (0, 0, "missing EC2 API key");
406 api_secret
= g_hash_table_lookup(args
,"api-secret");
408 api_secret
= (char *)prov
->password
;
409 if (!prov
->password
) {
410 error (0, 0, "missing EC2 API key");
415 cval
= g_hash_table_lookup(args
,"ami-cert");
417 ami_cert
= s3_init_tmpfile(cval
);
423 ami_cert
= get_provider_value(prov
,"ami-cert");
425 error (0, 0, "missing EC2 AMI cert");
430 kval
= g_hash_table_lookup(args
,"ami-key");
432 ami_key
= s3_init_tmpfile(kval
);
438 ami_key
= get_provider_value(prov
,"ami-key");
440 error (0, 0, "missing EC2 AMI key");
445 ami_uid
= g_hash_table_lookup(args
,"ami-uid");
447 ami_uid
= get_provider_value(prov
,"ami-uid");
449 error (0, 0, "missing EC2 AMI uid");
454 ami_bkt
= g_hash_table_lookup(args
,"ami-bkt");
456 ami_bkt
= ms
->bucket
;
460 * This is the point where we go from validation to execution. If we
461 * were double-forking so this could all be asynchronous, or for that
462 * matter to return an early 100-continue, this would probably be the
463 * place to do it. Even without that, we set the ami-id here so that
464 * the caller can know things are actually in progress.
466 sprintf(ami_id_buf
,"pending %lld",(long long)time(NULL
));
467 DPRINTF("temporary ami-id = \"%s\"\n",ami_id_buf
);
468 (void)meta_set_value(ms
->bucket
,ms
->key
,"ami-id",ami_id_buf
);
469 rc
= MHD_HTTP_INTERNAL_SERVER_ERROR
;
471 const char *cmd
= "dc-register-image";
473 argv
[argc
++] = ms
->bucket
;
474 argv
[argc
++] = ms
->key
;
475 argv
[argc
++] = api_key
;
476 argv
[argc
++] = api_secret
;
477 argv
[argc
++] = ami_cert
;
478 argv
[argc
++] = ami_key
;
479 argv
[argc
++] = ami_uid
;
480 argv
[argc
++] = ami_bkt
;
481 argv
[argc
++] = kernel
? kernel
: "_default_";
482 argv
[argc
++] = ramdisk
? ramdisk
: "_default_";
485 DPRINTF("api-key = %s\n",api_key
);
486 DPRINTF("api-secret = %s\n",api_secret
);
487 DPRINTF("ami-cert = %s\n",ami_cert
);
488 DPRINTF("ami-key = %s\n",ami_key
);
489 DPRINTF("ami-uid = %s\n",ami_uid
);
490 DPRINTF("ami-bkt = %s\n",ami_bkt
);
492 if (pipe(organ
) < 0) {
493 error (0, errno
, "pipe creation failed");
499 error (0, errno
, "fork failed");
506 (void)dup2(organ
[1],STDOUT_FILENO
);
507 (void)dup2(organ
[1],STDERR_FILENO
);
508 execvp(cmd
, (char* const*)argv
);
509 error (EXIT_FAILURE
, errno
, "failed run command %s", cmd
);
512 DPRINTF("waiting for child...\n");
513 if (waitpid(pid
,NULL
,0) < 0) {
514 error (0, errno
, "waitpid failed");
516 /* TBD: check identity/status from waitpid */
517 DPRINTF("...child exited\n");
520 fp
= fdopen(organ
[0],"r");
522 DPRINTF("could not open parent pipe\n");
526 while (fgets(buf
,sizeof(buf
)-1,fp
)) {
527 buf
[sizeof(buf
)-1] = '\0';
528 if (regexec(&s3_success_pat
,buf
,2,match
,0) == 0) {
529 buf
[match
[1].rm_eo
] = '\0';
530 DPRINTF("found AMI ID: %s\n",buf
+match
[1].rm_so
);
531 sprintf(ami_id_buf
,"OK %.60s",buf
+match
[1].rm_so
);
534 else if (regexec(&s3_failure_pat
,buf
,2,match
,0) == 0) {
535 buf
[match
[1].rm_eo
] = '\0';
536 DPRINTF("found error marker: %s\n",buf
+match
[1].rm_so
);
537 sprintf(ami_id_buf
,"failed %.56s",buf
+match
[1].rm_so
);
538 rc
= MHD_HTTP_INTERNAL_SERVER_ERROR
;
541 DPRINTF("ignoring line: <%s>\n",buf
);
548 * This is a bit tricky. If we found the cert in the HTTP request and
549 * succeeded in creating a temp file, then this condition will succeed.
550 * If we failed to create the temp file, or never found a cert
551 * anywhere, there will be no ami_cert to clean up. If we got a cert
552 * from the config, then ami_cert will be set but we'll (correctly)
553 * skip cleanup because cval is null.
555 if (cval
&& ami_cert
) {
557 free((char *)ami_cert
);
559 /* Same reasoning as above, with kval/ami_key. */
560 if (kval
&& ami_key
) {
562 free((char *)ami_key
);
564 (void)meta_set_value(ms
->bucket
,ms
->key
,"ami-id",ami_id_buf
);
569 /***** CURL-specific functions *****/
572 curl_init (provider_t
*prov
)
577 /* Start a CURL _producer_. */
579 curl_get_child (void * ctx
)
582 backend_thunk_t
*tp
= (backend_thunk_t
*)ctx
;
583 my_state
*ms
= tp
->parent
;
584 provider_t
*prov
= tp
->prov
;
587 curl
= curl_easy_init();
589 pipe_prod_siginit(&ms
->pipe
,-1);
590 return NULL
; /* TBD: flag error somehow */
592 if (ms
->from_master
) {
593 sprintf(fixed
,"http://%s:%u%s",
594 master_host
, master_port
, ms
->url
);
597 sprintf(fixed
,"http://%s:%u%s",
598 prov
->host
, prov
->port
, ms
->url
);
600 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
601 curl_easy_setopt(curl
,CURLOPT_WRITEFUNCTION
, http_get_prod
);
602 curl_easy_setopt(curl
,CURLOPT_WRITEDATA
,&ms
->pipe
);
603 pipe_prod_siginit(&ms
->pipe
,0);
605 curl_easy_perform(curl
);
606 curl_easy_getinfo(curl
,CURLINFO_RESPONSE_CODE
,&ms
->rc
);
607 pipe_prod_finish(&ms
->pipe
);
609 DPRINTF("producer exiting\n");
610 curl_easy_cleanup(curl
);
614 /* Start a CURL _consumer_. */
616 curl_put_child (void * ctx
)
618 pipe_private
*pp
= ctx
;
619 pipe_shared
*ps
= pp
->shared
;
620 my_state
*ms
= ps
->owner
;
621 provider_t
*prov
= pp
->prov
;
627 llen
= (curl_off_t
)MHD_SIZE_UNKNOWN
;
628 if (ms
->be_flags
& BACKEND_GET_SIZE
) {
629 clen
= MHD_lookup_connection_value(
630 ms
->conn
, MHD_HEADER_KIND
, "Content-Length");
632 llen
= strtoll(clen
,NULL
,10);
635 error (0, 0, "missing Content-Length");
639 curl
= curl_easy_init();
641 pipe_cons_siginit(ps
, -1);
643 return THREAD_FAILED
;
645 sprintf(fixed
,"http://%s:%u%s",prov
->host
,prov
->port
,
647 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
648 curl_easy_setopt(curl
,CURLOPT_UPLOAD
,1);
649 curl_easy_setopt(curl
,CURLOPT_INFILESIZE_LARGE
,llen
);
650 curl_easy_setopt(curl
,CURLOPT_READFUNCTION
,http_put_cons
);
651 curl_easy_setopt(curl
,CURLOPT_READDATA
,pp
);
652 pipe_cons_siginit(ps
, 0);
653 curl_easy_perform(curl
);
654 curl_easy_cleanup(curl
);
656 DPRINTF("%s returning\n",__func__
);
661 /* Start a CURL cache consumer. */
663 curl_cache_child (void * ctx
)
665 pipe_private
*pp
= ctx
;
666 pipe_shared
*ps
= pp
->shared
;
667 my_state
*ms
= ps
->owner
;
668 provider_t
*prov
= pp
->prov
;
672 char *my_url
= strdup(ms
->url
);
675 return THREAD_FAILED
;
678 curl
= curl_easy_init();
681 pipe_cons_siginit(ps
,-1);
682 return THREAD_FAILED
;
684 sprintf(fixed
,"http://%s:%u%s",prov
->host
,prov
->port
,
686 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
687 curl_easy_setopt(curl
,CURLOPT_UPLOAD
,1);
688 curl_easy_setopt(curl
,CURLOPT_INFILESIZE_LARGE
,
689 (curl_off_t
)MHD_SIZE_UNKNOWN
);
690 curl_easy_setopt(curl
,CURLOPT_READFUNCTION
,http_put_cons
);
691 curl_easy_setopt(curl
,CURLOPT_READDATA
,pp
);
692 curl_easy_perform(curl
);
693 curl_easy_cleanup(curl
);
695 slash
= index(my_url
+1,'/');
698 meta_got_copy(my_url
+1,slash
+1,me
);
706 curl_delete (const provider_t
*prov
, const char *bucket
, const char *key
,
715 curl
= curl_easy_init();
717 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
720 sprintf(fixed
,"http://%s:%u%s",prov
->host
,prov
->port
,url
);
721 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
722 curl_easy_setopt(curl
,CURLOPT_CUSTOMREQUEST
,"DELETE");
723 curl_easy_perform(curl
);
724 curl_easy_cleanup(curl
);
730 curl_bcreate (const provider_t
*prov
, const char *bucket
)
732 char addr
[ADDR_SIZE
];
736 chars
= snprintf(addr
,ADDR_SIZE
,"http://%s:%d/%s",
737 prov
->host
,prov
->port
,bucket
);
738 if (chars
>= ADDR_SIZE
) {
739 error(0,0,"path too long in %s",__func__
);
740 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
743 curl
= curl_easy_init();
745 error(0,errno
,"no memory in %s",__func__
);
746 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
748 curl_easy_setopt(curl
,CURLOPT_URL
,addr
);
749 curl_easy_setopt(curl
,CURLOPT_CUSTOMREQUEST
,"PUT");
750 curl_easy_perform(curl
);
751 curl_easy_cleanup(curl
);
756 * We can proxy through any number of CURL/HTTP warehouses, but the chain
757 * eventually has to terminate at an S3 back end.
761 curl_register (my_state
*ms
, const provider_t
*prov
, const char *next
,
766 struct curl_httppost
*first
= NULL
;
767 struct curl_httppost
*last
= NULL
;
768 char *kernel
= g_hash_table_lookup(args
,"kernel");
769 char *ramdisk
= g_hash_table_lookup(args
,"ramdisk");
772 DPRINTF("CURL register with next==NULL\n");
773 return MHD_HTTP_BAD_REQUEST
;
776 DPRINTF("*** PROXY registration request for %s/%s to %s (%s:%d)\n",
777 ms
->bucket
, ms
->key
, prov
->name
, prov
->host
, prov
->port
);
779 curl
= curl_easy_init();
781 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
783 sprintf(fixed
,"http://%s:%d/%s/%s",
784 prov
->host
,prov
->port
, ms
->bucket
, ms
->key
);
785 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
786 curl_formadd(&first
,&last
,
787 CURLFORM_COPYNAME
, "op",
788 CURLFORM_COPYCONTENTS
, "register",
790 curl_formadd(&first
,&last
,
791 CURLFORM_COPYNAME
, "site",
792 CURLFORM_COPYCONTENTS
, next
,
795 curl_formadd(&first
,&last
,
796 CURLFORM_COPYNAME
, "kernel",
797 CURLFORM_COPYCONTENTS
, kernel
,
801 curl_formadd(&first
,&last
,
802 CURLFORM_COPYNAME
, "ramdisk",
803 CURLFORM_COPYCONTENTS
, ramdisk
,
806 curl_easy_setopt(curl
,CURLOPT_HTTPPOST
,first
);
807 curl_easy_perform(curl
);
808 curl_easy_cleanup(curl
);
813 /***** CF-specific functions *****/
816 * TBD: refactor to maximize common code. Despite the de-duplication between
817 * this module and replica.c, there's still a lot more that could be done to
818 * combine xxx_yyy_child for xxx={http,cf} and yyy={put,cache}. A rough
821 * if xxx=cf, call CF-specific routine to add CF auth header
822 * do common curl setup and execution
823 * if yyy=cache, call meta_got_copy
825 * There might even be an opportunity to combine code for put and bucket
826 * create in some cases, since the only difference is the URL and the
827 * lack of a data transfer in the bucket-create case.
831 cf_writer (void *ptr ATTRIBUTE_UNUSED
, size_t size
, size_t nmemb
,
832 void *stream ATTRIBUTE_UNUSED
)
838 cf_header (void *ptr
, size_t size
, size_t nmemb
, void *stream
)
842 provider_t
*prov
= (provider_t
*)stream
;
844 next
= strtok_r(ptr
,":",&sctx
);
846 if (!strcasecmp(next
,"X-Storage-Url")) {
847 next
= strtok_r(NULL
," \n\r",&sctx
);
849 DPRINTF("got CF URL %s\n",next
);
850 /* NB: after this, original "host" is gone. */
851 free((char *)prov
->host
);
852 prov
->host
= strdup(next
);
855 else if (!strcasecmp(next
,"X-Storage-Token")) {
856 next
= strtok_r(NULL
," \n\r",&sctx
);
858 DPRINTF("got CF token %s\n",next
);
859 prov
->token
= strdup(next
);
866 static struct curl_slist
*
867 cf_add_token (struct curl_slist
*in_slist
, const char *token
)
870 char auth_hdr
[HEADER_SIZE
];
871 struct curl_slist
*out_slist
;
877 chars
= snprintf(auth_hdr
,HEADER_SIZE
,"X-Auth-Token: %s",token
);
878 if (chars
>= HEADER_SIZE
) {
879 error(0,0,"auth_token too long");
883 return curl_slist_append(NULL
,auth_hdr
);
887 cf_init (provider_t
*prov
)
890 char addr
[ADDR_SIZE
];
891 char auth_user
[HEADER_SIZE
];
892 char auth_key
[HEADER_SIZE
];
894 struct curl_slist
*slist
;
901 chars
= snprintf(addr
,ADDR_SIZE
,"https://%s:%u/v1.0",
902 prov
->host
, prov
->port
);
903 if (chars
>= ADDR_SIZE
) {
904 error(0,0,"API URL too long in %s",__func__
);
908 chars
= snprintf(auth_user
,HEADER_SIZE
,"X-Auth-User: %s",
910 if (chars
>= HEADER_SIZE
) {
911 error(0,0,"auth_user too long in %s",__func__
);
915 chars
= snprintf(auth_key
,HEADER_SIZE
,"X-Auth-Key: %s",
917 if (chars
>= HEADER_SIZE
) {
918 error(0,0,"auth_key too long in %s",__func__
);
922 curl
= curl_easy_init();
923 curl_easy_setopt(curl
,CURLOPT_URL
,addr
);
924 curl_easy_setopt(curl
,CURLOPT_WRITEFUNCTION
,cf_writer
);
925 curl_easy_setopt(curl
,CURLOPT_HEADERFUNCTION
,cf_header
);
926 curl_easy_setopt(curl
,CURLOPT_WRITEHEADER
,prov
);
927 slist
= curl_slist_append(NULL
,auth_user
);
928 slist
= curl_slist_append(slist
,auth_key
);
929 curl_easy_setopt(curl
,CURLOPT_HTTPHEADER
,slist
);
930 curl_easy_perform(curl
);
931 curl_easy_cleanup(curl
);
932 curl_slist_free_all(slist
);
934 DPRINTF("CF token = %s\n",prov
->token
);
937 /* Start a CloudFiles _producer_. */
939 cf_get_child (void * ctx
)
942 backend_thunk_t
*tp
= (backend_thunk_t
*)ctx
;
943 my_state
*ms
= tp
->parent
;
944 provider_t
*prov
= tp
->prov
;
946 struct curl_slist
*slist
= NULL
;
948 slist
= cf_add_token(slist
,prov
->token
);
950 return THREAD_FAILED
;
953 * Rackspace doesn't clearly document that you'll get
954 * 412 (Precondition Failed) if you omit this.
956 slist
= curl_slist_append(slist
,
957 "Content-Type: binary/octet-stream");
959 curl
= curl_easy_init();
961 pipe_prod_siginit(&ms
->pipe
,-1);
962 curl_slist_free_all(slist
);
963 return NULL
; /* TBD: flag error somehow */
965 sprintf(fixed
,"%s%s", prov
->host
, ms
->url
);
966 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
967 curl_easy_setopt(curl
,CURLOPT_WRITEFUNCTION
, http_get_prod
);
968 curl_easy_setopt(curl
,CURLOPT_WRITEDATA
,&ms
->pipe
);
969 curl_easy_setopt(curl
,CURLOPT_HTTPHEADER
,slist
);
970 pipe_prod_siginit(&ms
->pipe
,0);
972 curl_easy_perform(curl
);
973 curl_easy_getinfo(curl
,CURLINFO_RESPONSE_CODE
,&ms
->rc
);
974 pipe_prod_finish(&ms
->pipe
);
976 DPRINTF("producer exiting\n");
977 curl_easy_cleanup(curl
);
978 curl_slist_free_all(slist
);
982 /* Start a CloudFiles _consumer_. */
984 cf_put_child (void * ctx
)
986 pipe_private
*pp
= ctx
;
987 pipe_shared
*ps
= pp
->shared
;
988 my_state
*ms
= ps
->owner
;
989 provider_t
*prov
= pp
->prov
;
994 struct curl_slist
*slist
= NULL
;
996 slist
= cf_add_token(slist
,prov
->token
);
998 return THREAD_FAILED
;
1001 llen
= (curl_off_t
)MHD_SIZE_UNKNOWN
;
1002 if (ms
->be_flags
& BACKEND_GET_SIZE
) {
1003 clen
= MHD_lookup_connection_value(
1004 ms
->conn
, MHD_HEADER_KIND
, "Content-Length");
1006 llen
= strtoll(clen
,NULL
,10);
1009 error (0, 0, "missing Content-Length");
1013 curl
= curl_easy_init();
1015 pipe_cons_siginit(ps
, -1);
1017 curl_slist_free_all(slist
);
1018 return THREAD_FAILED
;
1020 sprintf(fixed
,"%s%s",prov
->host
,ms
->url
);
1021 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
1022 curl_easy_setopt(curl
,CURLOPT_UPLOAD
,1);
1023 curl_easy_setopt(curl
,CURLOPT_INFILESIZE_LARGE
,llen
);
1024 curl_easy_setopt(curl
,CURLOPT_READFUNCTION
,http_put_cons
);
1025 curl_easy_setopt(curl
,CURLOPT_READDATA
,pp
);
1026 curl_easy_setopt(curl
,CURLOPT_HTTPHEADER
,slist
);
1027 pipe_cons_siginit(ps
, 0);
1029 curl_easy_perform(curl
);
1030 curl_easy_getinfo(curl
,CURLINFO_RESPONSE_CODE
,&ms
->rc
);
1032 DPRINTF("%s returning\n",__func__
);
1033 curl_easy_cleanup(curl
);
1034 curl_slist_free_all(slist
);
1040 cf_delete (const provider_t
*prov
, const char *bucket
, const char *key
,
1046 char auth_hdr
[HEADER_SIZE
];
1048 struct curl_slist
*slist
= NULL
;
1050 slist
= cf_add_token(slist
,prov
->token
);
1052 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
1055 curl
= curl_easy_init();
1057 curl_slist_free_all(slist
);
1058 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
1061 sprintf(fixed
,"%s%s",prov
->host
,url
);
1062 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
1063 curl_easy_setopt(curl
,CURLOPT_CUSTOMREQUEST
,"DELETE");
1064 curl_easy_setopt(curl
,CURLOPT_HTTPHEADER
,slist
);
1066 curl_easy_perform(curl
);
1067 curl_easy_getinfo(curl
,CURLINFO_RESPONSE_CODE
,&rc
);
1068 DPRINTF("%s: rc = %ld\n",__func__
,rc
);
1070 curl_easy_cleanup(curl
);
1071 curl_slist_free_all(slist
);
1077 cf_null_reader (void *ptr
, size_t size
, size_t nmemb
, void *stream
)
1083 cf_bcreate (const provider_t
*prov
, const char *bucket
)
1088 char auth_hdr
[HEADER_SIZE
];
1090 struct curl_slist
*slist
= NULL
;
1092 slist
= cf_add_token(slist
,prov
->token
);
1094 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
1097 curl
= curl_easy_init();
1099 curl_slist_free_all(slist
);
1100 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
1102 sprintf(fixed
,"%s/%s",prov
->host
,bucket
);
1103 curl_easy_setopt(curl
,CURLOPT_URL
,fixed
);
1104 curl_easy_setopt(curl
,CURLOPT_UPLOAD
,1);
1105 curl_easy_setopt(curl
,CURLOPT_INFILESIZE_LARGE
,
1106 (curl_off_t
)MHD_SIZE_UNKNOWN
);
1107 curl_easy_setopt(curl
,CURLOPT_READFUNCTION
,cf_null_reader
);
1108 curl_easy_setopt(curl
,CURLOPT_HTTPHEADER
,slist
);
1110 curl_easy_perform(curl
);
1111 curl_easy_getinfo(curl
,CURLINFO_RESPONSE_CODE
,&rc
);
1112 DPRINTF("%s: rc = %ld\n",__func__
,rc
);
1114 DPRINTF("%s returning\n",__func__
);
1115 curl_easy_cleanup(curl
);
1116 curl_slist_free_all(slist
);
1120 /***** FS-specific functions *****/
1123 fs_init (provider_t
*prov
)
1125 DPRINTF("changing directory to %s\n",prov
->path
);
1126 if (chdir(prov
->path
) < 0) {
1127 error(0,errno
,"chdir failed, unsafe to continue");
1128 exit(!0); /* Value doesn't matter, as long as it's not zero. */
1132 /* Start an FS _producer_. */
1134 fs_get_child (void * ctx
)
1136 backend_thunk_t
*tp
= (backend_thunk_t
*)ctx
;
1137 my_state
*ms
= tp
->parent
;
1141 char *file
= ms
->url
+1;
1143 fd
= open(file
, O_RDONLY
);
1145 pipe_prod_siginit(&ms
->pipe
,-1);
1146 pipe_prod_finish(&ms
->pipe
);
1147 return THREAD_FAILED
;
1150 pipe_prod_siginit(&ms
->pipe
,0);
1153 bytes
= read(fd
,buf
,sizeof(buf
));
1156 error (0, errno
, "%s: read failed", file
);
1160 pipe_prod_signal(&ms
->pipe
,buf
,bytes
);
1164 pipe_prod_finish(&ms
->pipe
);
1166 DPRINTF("producer exiting\n");
1170 /* Start an FS _consumer_. */
1172 fs_put_child (void * ctx
)
1174 pipe_private
*pp
= ctx
;
1175 pipe_shared
*ps
= pp
->shared
;
1176 my_state
*ms
= ps
->owner
;
1180 char *file
= ms
->url
+1;
1182 fd
= open(file
,O_WRONLY
|O_CREAT
,0666);
1184 pipe_cons_siginit(ps
, errno
);
1186 return THREAD_FAILED
;
1189 pipe_cons_siginit(ps
, 0);
1191 while (pipe_cons_wait(pp
)) {
1195 ps
->data_ptr
+offset
,ps
->data_len
-offset
);
1198 error (0, errno
, "%s: write failed",
1201 pipe_cons_signal(pp
, errno
);
1205 } while (offset
< ps
->data_len
);
1206 pipe_cons_signal(pp
, 0);
1212 DPRINTF("%s returning\n",__func__
);
1218 fs_delete (const provider_t
*prov
, const char *bucket
, const char *key
,
1225 if (unlink(url
+1) < 0) {
1226 error (0, errno
, "%s: failed to unlink", url
+1);
1227 return MHD_HTTP_NOT_FOUND
;
1234 fs_bcreate (const provider_t
*prov
, const char *bucket
)
1238 DPRINTF("creating bucket %s\n",bucket
);
1240 if (mkdir(bucket
,0700) < 0) {
1241 error (0, errno
, "%s: failed to create directory", bucket
);
1242 return MHD_HTTP_INTERNAL_SERVER_ERROR
;
1248 /***** Function tables. ****/
1250 backend_func_tbl bad_func_tbl
= {
1261 backend_func_tbl s3_func_tbl
= {
1272 backend_func_tbl curl_func_tbl
= {
1283 backend_func_tbl cf_func_tbl
= {
1294 backend_func_tbl fs_func_tbl
= {