backend.c

   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 <getopt.h>
  20 #include <poll.h>
  21 #include <pthread.h>
  22 #include <regex.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 <sys/wait.h>
  32 #include <assert.h>
  33 #include <errno.h>
  34 #include <error.h>
  35
  36 #include <microhttpd.h>
  37 #include <curl/curl.h>
  38 #include <hstor.h>
  39 #include <glib.h>
  40
  41 #define GLOBALS_IMPL
  42 #include "iwh.h"
  43 #include "meta.h"
  44 #include "proxy.h"
  45 #include "template.h"
  46 #include "mpipe.h"
  47 #include "backend.h"
  48 #include "state_defs.h"
  49
  50 struct hstor_client     *hstor;
  51
  52 /***** Generic module stuff, not specific to one back end *****/
  53
  54 #define S3_IMAGE_PATTERN "^IMAGE[[:blank:]]+([^[:space:]]+)"
  55 #define S3_ERROR_PATTERN "^ERROR[[:blank:]]+([^[:space:]]+)"
  56
  57 regex_t s3_success_pat;
  58 regex_t s3_failure_pat;
  59 int     regex_ok = FALSE;
  60
  61 void
  62 backend_init (void)
  63 {
  64         regex_ok = TRUE;
  65
  66         if (regcomp(&s3_success_pat,S3_IMAGE_PATTERN,REG_EXTENDED) != 0){
  67                 DPRINTF("could not compile S3 success pattern\n");
  68                 regex_ok = FALSE;
  69         }
  70
  71         if (regcomp(&s3_failure_pat,S3_ERROR_PATTERN,REG_EXTENDED) != 0){
  72                 DPRINTF("could not compile S3 failure pattern\n");
  73                 regex_ok = FALSE;
  74         }
  75 }
  76
  77
  78 /***** Stub functions for unimplemented stuff. *****/
  79
  80 void
  81 bad_init (void)
  82 {
  83         DPRINTF("*** bad call to %s\n",__func__);
  84 }
  85
  86 void *
  87 bad_get_child (void * ctx)
  88 {
  89         (void)ctx;
  90
  91         DPRINTF("*** bad call to %s\n",__func__);
  92         return NULL;
  93 }
  94
  95 void *
  96 bad_put_child (void * ctx)
  97 {
  98         pipe_private    *pp     = ctx;
  99         pipe_shared     *ps     = pp->shared;
 100
 101         DPRINTF("*** bad call to %s\n",__func__);
 102         pipe_cons_siginit(ps, -1);
 103         free(pp);
 104         return THREAD_FAILED;
 105 }
 106
 107 void *
 108 bad_cache_child (void * ctx)
 109 {
 110         (void)ctx;
 111
 112         DPRINTF("*** bad call to %s\n",__func__);
 113         return NULL;
 114 }
 115
 116 int
 117 bad_delete (const char *bucket, const char *key, const char *url)
 118 {
 119         (void)bucket;
 120         (void)key;
 121         (void)url;
 122
 123         DPRINTF("*** bad call to %s\n",__func__);
 124         return MHD_HTTP_BAD_REQUEST;
 125 }
 126
 127 int
 128 bad_bcreate (const char *bucket)
 129 {
 130         (void)bucket;
 131
 132         DPRINTF("*** bad call to %s\n",__func__);
 133         return MHD_HTTP_NOT_IMPLEMENTED;
 134 }
 135
 136 int
 137 bad_register (my_state *ms, const provider_t *prov, const char *next,
 138               GHashTable *args)
 139 {
 140         (void)ms;
 141         (void)prov;
 142         (void)next;
 143         (void)args;
 144
 145         DPRINTF("*** bad call to %s\n",__func__);
 146         return MHD_HTTP_NOT_IMPLEMENTED;
 147 }
 148
 149 /***** Generic functions shared by the HTTP back ends. */
 150
 151 /* Invoked from S3/CURL/CF. */
 152 size_t
 153 http_get_prod (void *ptr, size_t size, size_t nmemb, void *stream)
 154 {
 155         size_t           total  = size * nmemb;
 156         pipe_shared     *ps     = stream;
 157
 158         DPRINTF("producer posting %zu bytes as %ld\n",total,ps->sequence+1);
 159         pipe_prod_signal(ps,ptr,total);
 160
 161         DPRINTF("producer finished chunk\n");
 162         return total;
 163 }
 164
 165 /* Invoked from S3/CURL/CF. */
 166 size_t
 167 http_put_cons (void *ptr, size_t size, size_t nmemb, void *stream)
 168 {
 169         size_t           total  = size * nmemb;
 170         pipe_private    *pp     = stream;
 171         pipe_shared     *ps     = pp->shared;
 172         size_t           done;
 173
 174         DPRINTF("consumer asked to read %zu\n",total);
 175
 176         pipe_cons_siginit(ps, 0);
 177
 178         if (!pipe_cons_wait(pp)) {
 179                 return 0;
 180         }
 181
 182         DPRINTF("consumer offset %zu into %zu\n",
 183                 pp->offset, ps->data_len);
 184         done = ps->data_len - pp->offset;
 185         if (done > total) {
 186                 done = total;
 187         }
 188         memcpy(ptr,ps->data_ptr+pp->offset,done);
 189         pp->offset += done;
 190         DPRINTF("consumer copied %zu, new offset %zu\n",
 191                 done, pp->offset);
 192         if (pp->offset == ps->data_len) {
 193                 DPRINTF("consumer finished chunk\n");
 194                 pipe_cons_signal(pp, 0);
 195         }
 196
 197         return done;
 198 }
 199
 200 /***** S3-specific functions *****/
 201
 202 void
 203 s3_init (void)
 204 {
 205         char svc_acc[128];
 206
 207         snprintf(svc_acc,sizeof(svc_acc),"%s:%u",
 208                 proxy_host,proxy_port);
 209         hstor = hstor_new(svc_acc,proxy_host,proxy_key,proxy_secret);
 210         if (hstor) {
 211                 if (verbose) {
 212                         hstor->verbose = 1;
 213                 }
 214         }
 215         else {
 216                 DPRINTF("could not create S3 client\n");
 217         }
 218 }
 219
 220 /* Start an S3 _producer_. */
 221 void *
 222 s3_get_child (void * ctx)
 223 {
 224         my_state        *ms     = ctx;
 225
 226         hstor_get(hstor,ms->bucket,ms->key,http_get_prod,&ms->pipe,0);
 227         /* TBD: check return value */
 228
 229         pipe_prod_finish(&ms->pipe);
 230
 231         DPRINTF("producer exiting\n");
 232         return NULL;
 233 }
 234
 235 /* Start an S3 _consumer_. */
 236 void *
 237 s3_put_child (void * ctx)
 238 {
 239         pipe_private    *pp     = ctx;
 240         pipe_shared     *ps     = pp->shared;
 241         my_state        *ms     = ps->owner;
 242         curl_off_t       llen;
 243         const char      *clen;
 244         bool             rcb;
 245
 246         clen = MHD_lookup_connection_value(
 247                 ms->conn, MHD_HEADER_KIND, "Content-Length");
 248         if (clen) {
 249                 llen = strtoll(clen,NULL,10);
 250         }
 251         else {
 252                 error (0, 0, "missing Content-Length");
 253                 llen = (curl_off_t)MHD_SIZE_UNKNOWN;
 254         }
 255
 256         rcb = hstor_put(hstor,ms->bucket,ms->key,http_put_cons,llen,pp,NULL);
 257         if (!rcb) {
 258                 DPRINTF("%s returning with error\n",__func__);
 259                 pipe_cons_siginit(ps, -1);
 260                 free(pp);
 261                 return THREAD_FAILED;
 262         }
 263
 264         DPRINTF("%s returning\n",__func__);
 265         free(pp);
 266         return NULL;
 267 }
 268
 269 int
 270 s3_delete (const char *bucket, const char *key, const char *url)
 271 {
 272         (void)url;
 273
 274         hstor_del(hstor,bucket,key);
 275         /* TBD: check return value */
 276
 277         return MHD_HTTP_OK;
 278 }
 279
 280 int
 281 s3_bcreate (const char *bucket)
 282 {
 283         DPRINTF("creating bucket %s\n",bucket);
 284
 285         if (!hstor_add_bucket(hstor,bucket)) {
 286                 DPRINTF("  bucket create failed\n");
 287                 return MHD_HTTP_INTERNAL_SERVER_ERROR;
 288         }
 289
 290         return MHD_HTTP_OK;
 291 }
 292
 293 char *
 294 s3_init_tmpfile (char *value)
 295 {
 296         char    *path;
 297         int      fd;
 298         size_t   len;
 299         ssize_t  written;
 300
 301         path = strdup("/tmp/iwtmp.XXXXXX");
 302         if (!path) {
 303                 return NULL;
 304         }
 305
 306         fd = mkstemp(path);
 307         if (fd < 0) {
 308                 error (0, errno, "%s: failed to create file from template", path);
 309                 free(path);
 310                 return NULL;
 311         }
 312
 313         len = strlen(value);
 314         if (len > 0) {
 315                 written = write(fd,value,len);
 316                 close(fd);
 317                 if (written != (ssize_t)len) {
 318                         if (written < 0) {
 319                                 error (0, errno, "failed to write to %s", path);
 320                         }
 321                         else {
 322                                 error (0, errno,
 323                                        "invalid write length %zd in %s",
 324                                        written, __func__);
 325                         }
 326                         unlink(path);
 327                         free(path);
 328                         return NULL;
 329                 }
 330         }
 331
 332         return path;
 333 }
 334
 335 int
 336 s3_register (my_state *ms, const provider_t *prov, const char *next,
 337              GHashTable *args)
 338 {
 339         char            *kernel         = g_hash_table_lookup(args,"kernel");
 340         char            *ramdisk        = g_hash_table_lookup(args,"ramdisk");
 341         char            *api_key;
 342         char            *api_secret;
 343         char            *ami_cert;
 344         char            *ami_key;
 345         char            *ami_uid;
 346         const char      *argv[12];
 347         int              argc = 0;
 348         pid_t            pid;
 349         int              organ[2];
 350         FILE            *fp;
 351         char             buf[1024];
 352         char            *cval   = NULL;
 353         char            *kval   = NULL;
 354         int              rc     = MHD_HTTP_BAD_REQUEST;
 355         char            *ami_bkt;
 356         char             ami_id_buf[64];
 357         regmatch_t       match[2];
 358
 359         if (!regex_ok) {
 360                 return MHD_HTTP_BAD_REQUEST;
 361         }
 362
 363         if (next) {
 364                 DPRINTF("S3 register with next!=NULL\n");
 365                 goto cleanup;
 366         }
 367
 368         DPRINTF("*** register %s/%s via %s (%s:%d)\n",
 369                 ms->bucket, ms->key, prov->name, prov->host, prov->port);
 370         if (kernel) {
 371                 DPRINTF("    (using kernel %s)\n",kernel);
 372         }
 373         if (ramdisk) {
 374                 DPRINTF("    (using ramdisk %s)\n",ramdisk);
 375         }
 376
 377         api_key = g_hash_table_lookup(args,"api-key");
 378         if (!api_key) {
 379                 api_key = (char *)prov->username;
 380                 if (!api_key) {
 381                         error (0, 0, "missing EC2 API key");
 382                         goto cleanup;
 383                 }
 384         }
 385
 386         api_secret = g_hash_table_lookup(args,"api-secret");
 387         if (!api_secret) {
 388                 api_secret = (char *)prov->password;
 389                 if (!prov->password) {
 390                         error (0, 0, "missing EC2 API key");
 391                         goto cleanup;
 392                 }
 393         }
 394
 395         cval = g_hash_table_lookup(args,"ami-cert");
 396         if (cval) {
 397                 ami_cert = s3_init_tmpfile(cval);
 398                 if (!ami_cert) {
 399                         goto cleanup;
 400                 }
 401         }
 402         else {
 403                 ami_cert = get_provider_value(prov->index,"ami-cert");
 404                 if (!ami_cert) {
 405                         error (0, 0, "missing EC2 AMI cert");
 406                         goto cleanup;
 407                 }
 408         }
 409
 410         kval = g_hash_table_lookup(args,"ami-key");
 411         if (kval) {
 412                 ami_key = s3_init_tmpfile(kval);
 413                 if (!ami_cert) {
 414                         goto cleanup;
 415                 }
 416         }
 417         else {
 418                 ami_key = get_provider_value(prov->index,"ami-key");
 419                 if (!ami_key) {
 420                         error (0, 0, "missing EC2 AMI key");
 421                         goto cleanup;
 422                 }
 423         }
 424
 425         ami_uid = g_hash_table_lookup(args,"ami-uid");
 426         if (!ami_uid) {
 427                 ami_uid = get_provider_value(prov->index,"ami-uid");
 428                 if (!ami_uid) {
 429                         error (0, 0, "missing EC2 AMI uid");
 430                         goto cleanup;
 431                 }
 432         }
 433
 434         ami_bkt = g_hash_table_lookup(args,"ami-bkt");
 435         if (!ami_bkt) {
 436                 ami_bkt = ms->bucket;
 437         }
 438
 439         /*
 440          * This is the point where we go from validation to execution.  If we
 441          * were double-forking so this could all be asynchronous, or for that
 442          * matter to return an early 100-continue, this would probably be the
 443          * place to do it.  Even without that, we set the ami-id here so that
 444          * the caller can know things are actually in progress.
 445          */
 446         sprintf(ami_id_buf,"pending %lld",(long long)time(NULL));
 447         DPRINTF("temporary ami-id = \"%s\"\n",ami_id_buf);
 448         (void)meta_set_value(ms->bucket,ms->key,"ami-id",ami_id_buf);
 449         rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
 450
 451         const char *cmd = "dc-register-image";
 452         argv[argc++] = cmd;
 453         argv[argc++] = ms->bucket;
 454         argv[argc++] = ms->key;
 455         argv[argc++] = api_key;
 456         argv[argc++] = api_secret;
 457         argv[argc++] = ami_cert;
 458         argv[argc++] = ami_key;
 459         argv[argc++] = ami_uid;
 460         argv[argc++] = ami_bkt;
 461         argv[argc++] = kernel ? kernel : "_default_";
 462         argv[argc++] = ramdisk ? ramdisk : "_default_";
 463         argv[argc] = NULL;
 464
 465         DPRINTF("api-key = %s\n",api_key);
 466         DPRINTF("api-secret = %s\n",api_secret);
 467         DPRINTF("ami-cert = %s\n",ami_cert);
 468         DPRINTF("ami-key = %s\n",ami_key);
 469         DPRINTF("ami-uid = %s\n",ami_uid);
 470         DPRINTF("ami-bkt = %s\n",ami_bkt);
 471
 472         if (pipe(organ) < 0) {
 473                 error (0, errno, "pipe creation failed");
 474                 goto cleanup;
 475         }
 476
 477         pid = fork();
 478         if (pid < 0) {
 479                 error (0, errno, "fork failed");
 480                 close(organ[0]);
 481                 close(organ[1]);
 482                 goto cleanup;
 483         }
 484
 485         if (pid == 0) {
 486                 (void)dup2(organ[1],STDOUT_FILENO);
 487                 (void)dup2(organ[1],STDERR_FILENO);
 488                 execvp(cmd, (char* const*)argv);
 489                 error (EXIT_FAILURE, errno, "failed run command %s", cmd);
 490         }
 491
 492         DPRINTF("waiting for child...\n");
 493         if (waitpid(pid,NULL,0) < 0) {
 494                 error (0, errno, "waitpid failed");
 495         }
 496         /* TBD: check identity/status from waitpid */
 497         DPRINTF("...child exited\n");
 498
 499         close(organ[1]);
 500         fp = fdopen(organ[0],"r");
 501         if (!fp) {
 502                 DPRINTF("could not open parent pipe\n");
 503                 close(organ[0]);
 504                 goto cleanup;
 505         }
 506         while (fgets(buf,sizeof(buf)-1,fp)) {
 507                 buf[sizeof(buf)-1] = '\0';
 508                 if (regexec(&s3_success_pat,buf,2,match,0) == 0) {
 509                         buf[match[1].rm_eo] = '\0';
 510                         DPRINTF("found AMI ID: %s\n",buf+match[1].rm_so);
 511                         sprintf(ami_id_buf,"OK %.60s",buf+match[1].rm_so);
 512                         rc = MHD_HTTP_OK;
 513                 }
 514                 else if (regexec(&s3_failure_pat,buf,2,match,0) == 0) {
 515                         buf[match[1].rm_eo] = '\0';
 516                         DPRINTF("found error marker: %s\n",buf+match[1].rm_so);
 517                         sprintf(ami_id_buf,"failed %.56s",buf+match[1].rm_so);
 518                         rc = MHD_HTTP_INTERNAL_SERVER_ERROR;
 519                 }
 520                 else {
 521                         DPRINTF("ignoring line: <%s>\n",buf);
 522                 }
 523         }
 524         fclose(fp);
 525
 526 cleanup:
 527         /*
 528          * This is a bit tricky.  If we found the cert in the HTTP request and
 529          * succeeded in creating a temp file, then this condition will succeed.
 530          * If we failed to create the temp file, or never found a cert
 531          * anywhere, there will be no ami_cert to clean up.  If we got a cert
 532          * from the config, then ami_cert will be set but we'll (correctly)
 533          * skip cleanup because cval is null.
 534          */
 535         if (cval && ami_cert) {
 536                 unlink(ami_cert);
 537                 free(ami_cert);
 538         }
 539         /* Same reasoning as above, with kval/ami_key. */
 540         if (kval && ami_key) {
 541                 unlink(ami_key);
 542                 free(ami_key);
 543         }
 544         (void)meta_set_value(ms->bucket,ms->key,"ami-id",ami_id_buf);
 545
 546         return rc;
 547 }
 548
 549 /***** CURL-specific functions *****/
 550
 551 void
 552 curl_init (void)
 553 {
 554 }
 555
 556 /* Start a CURL _producer_. */
 557 void *
 558 curl_get_child (void * ctx)
 559 {
 560         char             fixed[1024];
 561         my_state        *ms     = ctx;
 562
 563         ms->curl = curl_easy_init();
 564         if (!ms->curl) {
 565                 return NULL;    /* TBD: flag error somehow */
 566         }
 567         ms->cleanup |= CLEANUP_CURL;
 568         if (ms->from_master) {
 569                 sprintf(fixed,"http://%s:%u%s",
 570                         master_host, master_port, ms->url);
 571         }
 572         else {
 573                 sprintf(fixed,"http://%s:%u%s",
 574                         proxy_host, proxy_port, ms->url);
 575         }
 576         curl_easy_setopt(ms->curl,CURLOPT_URL,fixed);
 577         curl_easy_setopt(ms->curl,CURLOPT_WRITEFUNCTION,
 578                          http_get_prod);
 579         curl_easy_setopt(ms->curl,CURLOPT_WRITEDATA,&ms->pipe);
 580         curl_easy_perform(ms->curl);
 581         curl_easy_getinfo(ms->curl,CURLINFO_RESPONSE_CODE,&ms->rc);
 582
 583         pipe_prod_finish(&ms->pipe);
 584
 585         DPRINTF("producer exiting\n");
 586         return NULL;
 587 }
 588
 589 /* Start a CURL _consumer_. */
 590 void *
 591 curl_put_child (void * ctx)
 592 {
 593         pipe_private    *pp     = ctx;
 594         pipe_shared     *ps     = pp->shared;
 595         my_state        *ms     = ps->owner;
 596         curl_off_t       llen;
 597         char             fixed[1024];
 598         CURL            *curl;
 599         const char      *clen;
 600
 601         clen = MHD_lookup_connection_value(
 602                 ms->conn, MHD_HEADER_KIND, "Content-Length");
 603         if (clen) {
 604                 llen = strtoll(clen,NULL,10);
 605         }
 606         else {
 607                 error (0, 0, "missing Content-Length");
 608                 llen = (curl_off_t)MHD_SIZE_UNKNOWN;
 609         }
 610
 611         curl = curl_easy_init();
 612         if (!curl) {
 613                 pipe_cons_siginit(ps, -1);
 614                 free(pp);
 615                 return THREAD_FAILED;
 616         }
 617         sprintf(fixed,"http://%s:%u%s",proxy_host,proxy_port,
 618                 ms->url);
 619         curl_easy_setopt(curl,CURLOPT_URL,fixed);
 620         curl_easy_setopt(curl,CURLOPT_UPLOAD,1);
 621         curl_easy_setopt(curl,CURLOPT_INFILESIZE_LARGE,llen);
 622         curl_easy_setopt(curl,CURLOPT_READFUNCTION,http_put_cons);
 623         curl_easy_setopt(curl,CURLOPT_READDATA,pp);
 624         curl_easy_perform(curl);
 625         curl_easy_cleanup(curl);
 626
 627         DPRINTF("%s returning\n",__func__);
 628         free(pp);
 629         return NULL;
 630 }
 631
 632 /* Start a CURL cache consumer. */
 633 void *
 634 curl_cache_child (void * ctx)
 635 {
 636         pipe_private    *pp     = ctx;
 637         pipe_shared     *ps     = pp->shared;
 638         my_state        *ms     = ps->owner;
 639         char             fixed[1024];
 640         CURL            *curl;
 641         char            *slash;
 642         char            *my_url = strdup(ms->url);
 643
 644         if (!my_url) {
 645                 return THREAD_FAILED;
 646         }
 647
 648         curl = curl_easy_init();
 649         if (!curl) {
 650                 free(my_url);
 651                 return THREAD_FAILED;
 652         }
 653         sprintf(fixed,"http://%s:%u%s",proxy_host,proxy_port,
 654                 ms->url);
 655         curl_easy_setopt(curl,CURLOPT_URL,fixed);
 656         curl_easy_setopt(curl,CURLOPT_UPLOAD,1);
 657         curl_easy_setopt(curl,CURLOPT_INFILESIZE_LARGE,
 658                 (curl_off_t)MHD_SIZE_UNKNOWN);
 659         curl_easy_setopt(curl,CURLOPT_READFUNCTION,http_put_cons);
 660         curl_easy_setopt(curl,CURLOPT_READDATA,pp);
 661         curl_easy_perform(curl);
 662         curl_easy_cleanup(curl);
 663
 664         slash = index(my_url+1,'/');
 665         if (slash) {
 666                 *slash = '\0';
 667                 meta_got_copy(my_url+1,slash+1,me);
 668         }
 669
 670         free(my_url);
 671         return NULL;
 672 }
 673
 674 int
 675 curl_delete (const char *bucket, const char *key, const char *url)
 676 {
 677         CURL                    *curl;
 678         char                     fixed[1024];
 679
 680         (void)bucket;
 681         (void)key;
 682
 683         curl = curl_easy_init();
 684         if (!curl) {
 685                 return MHD_HTTP_INTERNAL_SERVER_ERROR;
 686         }
 687
 688         sprintf(fixed,"http://%s:%u%s",proxy_host,proxy_port,url);
 689         curl_easy_setopt(curl,CURLOPT_URL,fixed);
 690         curl_easy_setopt(curl,CURLOPT_CUSTOMREQUEST,"DELETE");
 691         curl_easy_perform(curl);
 692         curl_easy_cleanup(curl);
 693
 694         return MHD_HTTP_OK;
 695 }
 696
 697 int
 698 curl_bcreate (const char *bucket)
 699 {
 700         (void)bucket;
 701
 702         DPRINTF("cannot create bucket in non-S3 mode\n");
 703         /* TBD: pretend this works for testing, fix for release
 704         rc = MHD_HTTP_NOT_IMPLEMENTED;
 705         */
 706         return MHD_HTTP_OK;
 707 }
 708
 709 /*
 710  * We can proxy through any number of CURL/HTTP warehouses, but the chain
 711  * eventually has to terminate at an S3 back end.
 712  */
 713
 714 int
 715 curl_register (my_state *ms, const provider_t *prov, const char *next,
 716                GHashTable *args)
 717 {
 718         char                     fixed[1024];
 719         CURL                    *curl;
 720         struct curl_httppost    *first  = NULL;
 721         struct curl_httppost    *last   = NULL;
 722         char    *kernel         = g_hash_table_lookup(args,"kernel");
 723         char    *ramdisk        = g_hash_table_lookup(args,"ramdisk");
 724
 725         if (!next) {
 726                 DPRINTF("CURL register with next==NULL\n");
 727                 return MHD_HTTP_BAD_REQUEST;
 728         }
 729
 730         DPRINTF("*** PROXY registration request for %s/%s to %s (%s:%d)\n",
 731                 ms->bucket, ms->key, prov->name, prov->host, prov->port);
 732
 733         curl = curl_easy_init();
 734         if (!curl) {
 735                 return MHD_HTTP_INTERNAL_SERVER_ERROR;
 736         }
 737         sprintf(fixed,"http://%s:%d/%s/%s",
 738                 prov->host,prov->port, ms->bucket, ms->key);
 739         curl_easy_setopt(curl,CURLOPT_URL,fixed);
 740         curl_formadd(&first,&last,
 741                 CURLFORM_COPYNAME, "op",
 742                 CURLFORM_COPYCONTENTS, "register",
 743                 CURLFORM_END);
 744         curl_formadd(&first,&last,
 745                 CURLFORM_COPYNAME, "site",
 746                 CURLFORM_COPYCONTENTS, next,
 747                 CURLFORM_END);
 748         if (kernel) {
 749                 curl_formadd(&first,&last,
 750                         CURLFORM_COPYNAME, "kernel",
 751                         CURLFORM_COPYCONTENTS, kernel,
 752                         CURLFORM_END);
 753         }
 754         if (ramdisk) {
 755                 curl_formadd(&first,&last,
 756                         CURLFORM_COPYNAME, "ramdisk",
 757                         CURLFORM_COPYCONTENTS, ramdisk,
 758                         CURLFORM_END);
 759         }
 760         curl_easy_setopt(curl,CURLOPT_HTTPPOST,first);
 761         curl_easy_perform(curl);
 762         curl_easy_cleanup(curl);
 763
 764         return MHD_HTTP_OK;
 765 }
 766
 767 /***** CF-specific functions (TBD) *****/
 768
 769 /***** FS-specific functions *****/
 770
 771 void
 772 fs_init (void)
 773 {
 774         DPRINTF("changing directory to %s\n",local_path);
 775         if (chdir(local_path) < 0) {
 776                 error(0,errno,"chdir failed, unsafe to continue");
 777                 exit(!0); /* Value doesn't matter, as long as it's not zero. */
 778         }
 779 }
 780
 781 /* Start an FS _producer_. */
 782 void *
 783 fs_get_child (void * ctx)
 784 {
 785         my_state        *ms     = ctx;
 786         int              fd;
 787         char             buf[1<<16];
 788         ssize_t          bytes;
 789         char            *file = ms->url+1;
 790
 791         fd = open(file, O_RDONLY);
 792         if (fd < 0) {
 793                 return THREAD_FAILED;
 794         }
 795
 796         for (;;) {
 797                 bytes = read(fd,buf,sizeof(buf));
 798                 if (bytes <= 0) {
 799                         if (bytes < 0) {
 800                                 error (0, errno, "%s: read failed", file);
 801                         }
 802                         break;
 803                 }
 804                 pipe_prod_signal(&ms->pipe,buf,bytes);
 805         }
 806
 807         close(fd);
 808         pipe_prod_finish(&ms->pipe);
 809
 810         DPRINTF("producer exiting\n");
 811         return NULL;
 812 }
 813
 814 /* Start an FS _consumer_. */
 815 void *
 816 fs_put_child (void * ctx)
 817 {
 818         pipe_private    *pp     = ctx;
 819         pipe_shared     *ps     = pp->shared;
 820         my_state        *ms     = ps->owner;
 821         int              fd;
 822         ssize_t          bytes;
 823         size_t           offset;
 824         char            *file = ms->url+1;
 825
 826         fd = open(file,O_WRONLY|O_CREAT,0666);
 827         if (fd < 0) {
 828                 pipe_cons_siginit(ps, errno);
 829                 free(pp);
 830                 return THREAD_FAILED;
 831         }
 832
 833         pipe_cons_siginit(ps, 0);
 834
 835         while (pipe_cons_wait(pp)) {
 836                 offset = 0;
 837                 do {
 838                         bytes = write(fd,
 839                                 ps->data_ptr+offset,ps->data_len-offset);
 840                         if (bytes <= 0) {
 841                                 if (bytes < 0) {
 842                                         error (0, errno, "%s: write failed",
 843                                                file);
 844                                 }
 845                                 pipe_cons_signal(pp, errno);
 846                                 goto done;
 847                         }
 848                         offset += bytes;
 849                 } while (offset < ps->data_len);
 850                 pipe_cons_signal(pp, 0);
 851         }
 852
 853 done:
 854         close(fd);
 855
 856         DPRINTF("%s returning\n",__func__);
 857         free(pp);
 858         return NULL;
 859 }
 860
 861 int
 862 fs_delete (const char *bucket, const char *key, const char *url)
 863 {
 864         (void)bucket;
 865         (void)key;
 866
 867         if (unlink(url+1) < 0) {
 868                 error (0, errno, "%s: failed to unlink", url+1);
 869                 return MHD_HTTP_NOT_FOUND;
 870         }
 871
 872         return MHD_HTTP_OK;
 873 }
 874
 875 int
 876 fs_bcreate (const char *bucket)
 877 {
 878         DPRINTF("creating bucket %s\n",bucket);
 879
 880         if (mkdir(bucket,0700) < 0) {
 881                 error (0, errno, "%s: failed to create directory", bucket);
 882                 return MHD_HTTP_INTERNAL_SERVER_ERROR;
 883         }
 884
 885         return MHD_HTTP_OK;
 886 }
 887
 888
 889 /***** Function tables. ****/
 890
 891 backend_func_tbl bad_func_tbl = {
 892         "uninitialized",
 893         bad_init,
 894         bad_get_child,
 895         bad_put_child,
 896         bad_cache_child,
 897         bad_delete,
 898         bad_bcreate,
 899         bad_register,
 900 };
 901
 902 backend_func_tbl s3_func_tbl = {
 903         "S3",
 904         s3_init,
 905         s3_get_child,
 906         s3_put_child,
 907         bad_cache_child,
 908         s3_delete,
 909         s3_bcreate,
 910         s3_register,
 911 };
 912
 913 backend_func_tbl curl_func_tbl = {
 914         "HTTP",
 915         curl_init,
 916         curl_get_child,
 917         curl_put_child,
 918         curl_cache_child,
 919         curl_delete,
 920         curl_bcreate,
 921         curl_register,
 922 };
 923
 924 backend_func_tbl fs_func_tbl = {
 925         "FS",
 926         fs_init,
 927         fs_get_child,
 928         fs_put_child,
 929         bad_cache_child,
 930         fs_delete,
 931         fs_bcreate,
 932         bad_register,
 933 };