plugins/out_s3/s3.c
static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk,
                       struct multipart_upload *m_upload,
                       char *body, size_t body_size,
                       const char *tag, int tag_len)
{
    int init_upload = FLB_FALSE;
    int complete_upload = FLB_FALSE;
    int size_check = FLB_FALSE;
    int part_num_check = FLB_FALSE;
    int timeout_check = FLB_FALSE;
    time_t create_time;
    int ret;

    if (ctx->use_put_object == FLB_TRUE) {
        goto put_object;
    }

    if (s3_plugin_under_test() == FLB_TRUE) {
        init_upload = FLB_TRUE;
        complete_upload = FLB_TRUE;
        if (ctx->use_put_object == FLB_TRUE) {
            goto put_object;
        }
        else {
            goto multipart;
        }
    }

    if (m_upload == NULL) {
        if (chunk != NULL && time(NULL) > (chunk->create_time + ctx->upload_timeout)) {
            /* timeout already reached, just PutObject */
            goto put_object;
        }
        else if (body_size >= ctx->file_size) {
            /* already big enough, just use PutObject API */
            goto put_object;
        }
        else if(body_size > MIN_CHUNKED_UPLOAD_SIZE) {
            init_upload = FLB_TRUE;
            goto multipart;
        }
        else {
            goto put_object;
        }
    }
    else {
        /* existing upload */
        if (body_size < MIN_CHUNKED_UPLOAD_SIZE) {
            complete_upload = FLB_TRUE;
        }

        goto multipart;
    }

put_object:

    /*
     * remove chunk from buffer list- needed for async http so that the
     * same chunk won't be sent more than once
     */
    if (chunk) {
        create_time = chunk->create_time;
    }
    else {
        create_time = time(NULL);
    }

    ret = s3_put_object(ctx, tag, create_time, body, body_size);
    if (ret < 0) {
        /* re-add chunk to list */
        if (chunk) {
            s3_store_file_unlock(chunk);
            chunk->failures += 1;
        }
        return FLB_RETRY;
    }

    /* data was sent successfully- delete the local buffer */
    if (chunk) {
        s3_store_file_delete(ctx, chunk);
    }
    return FLB_OK;

multipart:

    if (init_upload == FLB_TRUE) {
        m_upload = create_upload(ctx, tag, tag_len);
        if (!m_upload) {
            flb_plg_error(ctx->ins, "Could not find or create upload for tag %s", tag);
            if (chunk) {
                s3_store_file_unlock(chunk);
            }
            return FLB_RETRY;
        }
    }

    if (m_upload->upload_state == MULTIPART_UPLOAD_STATE_NOT_CREATED) {
        ret = create_multipart_upload(ctx, m_upload);
        if (ret < 0) {
            flb_plg_error(ctx->ins, "Could not initiate multipart upload");
            if (chunk) {
                s3_store_file_unlock(chunk);
            }
            return FLB_RETRY;
        }
        m_upload->upload_state = MULTIPART_UPLOAD_STATE_CREATED;
    }

    ret = upload_part(ctx, m_upload, body, body_size);
    if (ret < 0) {
        m_upload->upload_errors += 1;
        /* re-add chunk to list */
        if (chunk) {
            s3_store_file_unlock(chunk);
            chunk->failures += 1;
        }
        return FLB_RETRY;
    }
    m_upload->part_number += 1;

    /* data was sent successfully- delete the local buffer */
    if (chunk) {
        s3_store_file_delete(ctx, chunk);
        chunk = NULL;
    }

    if (m_upload->bytes >= ctx->file_size) {
        size_check = FLB_TRUE;
        flb_plg_info(ctx->ins, "Will complete upload for %s because uploaded data is greater"
                     " than size set by total_file_size", m_upload->s3_key);
    }
    if (m_upload->part_number >= 10000) {
        part_num_check = FLB_TRUE;
        flb_plg_info(ctx->ins, "Will complete upload for %s because 10,000 chunks "
                     "(the API limit) have been uploaded", m_upload->s3_key);
    }
    if (time(NULL) > (m_upload->init_time + ctx->upload_timeout)) {
        timeout_check = FLB_TRUE;
        flb_plg_info(ctx->ins, "Will complete upload for %s because upload_timeout"
                     " has elapsed", m_upload->s3_key);
    }
    if (size_check || part_num_check || timeout_check) {
        complete_upload = FLB_TRUE;
    }

    if (complete_upload == FLB_TRUE) {
        /* mark for completion- the upload timer will handle actual completion */
        m_upload->upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS;
    }

    return FLB_OK;
}
/*
 * return value is one of FLB_OK, FLB_RETRY, FLB_ERROR
 *
 * Chunk is allowed to be NULL
 */
return value is one of FLB_OK, FLB_RETRY, FLB_ERROR
Chunk is allowed to be NULL
return value is one of FLB_OK FLB_RETRY FLB_ERROR Chunk is allowed to be NULL
static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk,
                       struct multipart_upload *m_upload,
                       char *body, size_t body_size,
                       const char *tag, int tag_len)
{
    int init_upload = FLB_FALSE;
    int complete_upload = FLB_FALSE;
    int size_check = FLB_FALSE;
    int part_num_check = FLB_FALSE;
    int timeout_check = FLB_FALSE;
    time_t create_time;
    int ret;
    if (ctx->use_put_object == FLB_TRUE) {
        goto put_object;
    }
    if (s3_plugin_under_test() == FLB_TRUE) {
        init_upload = FLB_TRUE;
        complete_upload = FLB_TRUE;
        if (ctx->use_put_object == FLB_TRUE) {
            goto put_object;
        }
        else {
            goto multipart;
        }
    }
    if (m_upload == NULL) {
        if (chunk != NULL && time(NULL) > (chunk->create_time + ctx->upload_timeout)) {
            goto put_object;
        }
        else if (body_size >= ctx->file_size) {
            goto put_object;
        }
        else if(body_size > MIN_CHUNKED_UPLOAD_SIZE) {
            init_upload = FLB_TRUE;
            goto multipart;
        }
        else {
            goto put_object;
        }
    }
    else {
        if (body_size < MIN_CHUNKED_UPLOAD_SIZE) {
            complete_upload = FLB_TRUE;
        }
        goto multipart;
    }
put_object:
    if (chunk) {
        create_time = chunk->create_time;
    }
    else {
        create_time = time(NULL);
    }
    ret = s3_put_object(ctx, tag, create_time, body, body_size);
    if (ret < 0) {
        if (chunk) {
            s3_store_file_unlock(chunk);
            chunk->failures += 1;
        }
        return FLB_RETRY;
    }
    if (chunk) {
        s3_store_file_delete(ctx, chunk);
    }
    return FLB_OK;
multipart:
    if (init_upload == FLB_TRUE) {
        m_upload = create_upload(ctx, tag, tag_len);
        if (!m_upload) {
            flb_plg_error(ctx->ins, "Could not find or create upload for tag %s", tag);
            if (chunk) {
                s3_store_file_unlock(chunk);
            }
            return FLB_RETRY;
        }
    }
    if (m_upload->upload_state == MULTIPART_UPLOAD_STATE_NOT_CREATED) {
        ret = create_multipart_upload(ctx, m_upload);
        if (ret < 0) {
            flb_plg_error(ctx->ins, "Could not initiate multipart upload");
            if (chunk) {
                s3_store_file_unlock(chunk);
            }
            return FLB_RETRY;
        }
        m_upload->upload_state = MULTIPART_UPLOAD_STATE_CREATED;
    }
    ret = upload_part(ctx, m_upload, body, body_size);
    if (ret < 0) {
        m_upload->upload_errors += 1;
        if (chunk) {
            s3_store_file_unlock(chunk);
            chunk->failures += 1;
        }
        return FLB_RETRY;
    }
    m_upload->part_number += 1;
    if (chunk) {
        s3_store_file_delete(ctx, chunk);
        chunk = NULL;
    }
    if (m_upload->bytes >= ctx->file_size) {
        size_check = FLB_TRUE;
        flb_plg_info(ctx->ins, "Will complete upload for %s because uploaded data is greater"
                     " than size set by total_file_size", m_upload->s3_key);
    }
    if (m_upload->part_number >= 10000) {
        part_num_check = FLB_TRUE;
        flb_plg_info(ctx->ins, "Will complete upload for %s because 10,000 chunks "
                     "(the API limit) have been uploaded", m_upload->s3_key);
    }
    if (time(NULL) > (m_upload->init_time + ctx->upload_timeout)) {
        timeout_check = FLB_TRUE;
        flb_plg_info(ctx->ins, "Will complete upload for %s because upload_timeout"
                     " has elapsed", m_upload->s3_key);
    }
    if (size_check || part_num_check || timeout_check) {
        complete_upload = FLB_TRUE;
    }
    if (complete_upload == FLB_TRUE) {
        m_upload->upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS;
    }
    return FLB_OK;
}
static int upload_data ( struct flb_s3 * ctx , struct s3_file * chunk , struct multipart_upload * m_upload , char * body , size_t body_size , const char * tag , int tag_len ) { int init_upload = FLB_FALSE ; int complete_upload = FLB_FALSE ; int size_check = FLB_FALSE ; int part_num_check = FLB_FALSE ; int timeout_check = FLB_FALSE ; time_t create_time ; int ret ; if ( ctx -> use_put_object == FLB_TRUE ) { goto put_object ; } if ( s3_plugin_under_test ( ) == FLB_TRUE ) { init_upload = FLB_TRUE ; complete_upload = FLB_TRUE ; if ( ctx -> use_put_object == FLB_TRUE ) { goto put_object ; } else { goto multipart ; } } if ( m_upload == NULL ) { if ( chunk != NULL && time ( NULL ) > ( chunk -> create_time + ctx -> upload_timeout ) ) { goto put_object ; } else if ( body_size >= ctx -> file_size ) { goto put_object ; } else if ( body_size > MIN_CHUNKED_UPLOAD_SIZE ) { init_upload = FLB_TRUE ; goto multipart ; } else { goto put_object ; } } else { if ( body_size < MIN_CHUNKED_UPLOAD_SIZE ) { complete_upload = FLB_TRUE ; } goto multipart ; } put_object : if ( chunk ) { create_time = chunk -> create_time ; } else { create_time = time ( NULL ) ; } ret = s3_put_object ( ctx , tag , create_time , body , body_size ) ; if ( ret < 0 ) { if ( chunk ) { s3_store_file_unlock ( chunk ) ; chunk -> failures += 1 ; } return FLB_RETRY ; } if ( chunk ) { s3_store_file_delete ( ctx , chunk ) ; } return FLB_OK ; multipart : if ( init_upload == FLB_TRUE ) { m_upload = create_upload ( ctx , tag , tag_len ) ; if ( ! m_upload ) { flb_plg_error ( ctx -> ins , " " , tag ) ; if ( chunk ) { s3_store_file_unlock ( chunk ) ; } return FLB_RETRY ; } } if ( m_upload -> upload_state == MULTIPART_UPLOAD_STATE_NOT_CREATED ) { ret = create_multipart_upload ( ctx , m_upload ) ; if ( ret < 0 ) { flb_plg_error ( ctx -> ins , " " ) ; if ( chunk ) { s3_store_file_unlock ( chunk ) ; } return FLB_RETRY ; } m_upload -> upload_state = MULTIPART_UPLOAD_STATE_CREATED ; } ret = upload_part ( ctx , m_upload , body , body_size ) ; if ( ret < 0 ) { m_upload -> upload_errors += 1 ; if ( chunk ) { s3_store_file_unlock ( chunk ) ; chunk -> failures += 1 ; } return FLB_RETRY ; } m_upload -> part_number += 1 ; if ( chunk ) { s3_store_file_delete ( ctx , chunk ) ; chunk = NULL ; } if ( m_upload -> bytes >= ctx -> file_size ) { size_check = FLB_TRUE ; flb_plg_info ( ctx -> ins , " " " " , m_upload -> s3_key ) ; } if ( m_upload -> part_number >= 10000 ) { part_num_check = FLB_TRUE ; flb_plg_info ( ctx -> ins , " " " " , m_upload -> s3_key ) ; } if ( time ( NULL ) > ( m_upload -> init_time + ctx -> upload_timeout ) ) { timeout_check = FLB_TRUE ; flb_plg_info ( ctx -> ins , " " " " , m_upload -> s3_key ) ; } if ( size_check || part_num_check || timeout_check ) { complete_upload = FLB_TRUE ; } if ( complete_upload == FLB_TRUE ) { m_upload -> upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS ; } return FLB_OK ; }
return value is one of FLB_OK, FLB_RETRY, FLB_ERROR
 Chunk is allowed to be NULL
return value is one of FLB_OK FLB_RETRY FLB_ERROR Chunk is allowed to be NULL
/* timeout already reached, just PutObject */ /* already big enough, just use PutObject API */ /* existing upload */ /*
     * remove chunk from buffer list- needed for async http so that the
     * same chunk won't be sent more than once
     */ /* re-add chunk to list */ /* data was sent successfully- delete the local buffer */ /* re-add chunk to list */ /* data was sent successfully- delete the local buffer */ /* mark for completion- the upload timer will handle actual completion */