plugins/slice/client.cc

/** @file
  Licensed to the Apache Software Foundation (ASF) under one
  or more contributor license agreements.  See the NOTICE file
  distributed with this work for additional information
  regarding copyright ownership.  The ASF licenses this file
  to you under the Apache License, Version 2.0 (the
  "License"); you may not use this file except in compliance
  with the License.  You may obtain a copy of the License at

      http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License.
 */

#include "client.h"

#include "Config.h"
#include "server.h"
#include "util.h"

#include "swoc/TextView.h"

#include <cctype>
#include <cinttypes>
#include <limits>

namespace
{
// Miss bound for this purge: the request header when usable, else the config value.
int
purge_miss_bound(HttpHeader const &header, Config *const conf)
{
  char probestr[64];
  int  probelen = sizeof(probestr);

  if (!header.valueForKey(conf->m_purge_probe_header.data(), conf->m_purge_probe_header.size(), probestr, &probelen)) {
    return conf->m_purge_probe_blocks;
  }

  swoc::TextView value{probestr, static_cast<size_t>(probelen)};
  // isspace is only defined for values representable as unsigned char
  value.trim_if([](char c) { return 0 != isspace(static_cast<unsigned char>(c)); });

  swoc::TextView parsed;
  intmax_t const blocks = swoc::svtoi(value, &parsed, 10);

  // parsed must cover the whole value: "8abc" is a mistake, not eight blocks
  if (parsed.size() != value.size() || blocks <= 0 || std::numeric_limits<int>::max() < blocks) {
    // paced: the value is client supplied, so a bad one repeats as fast as requests arrive
    if (conf->canLogError()) {
      ERROR_LOG("Ignoring invalid %.*s value '%.*s'", static_cast<int>(conf->m_purge_probe_header.size()),
                conf->m_purge_probe_header.data(), probelen, probestr);
    }
    return conf->m_purge_probe_blocks;
  }

  return static_cast<int>(blocks);
}

} // namespace

// this is called once per transaction when the client sends a req header
bool
handle_client_req(TSCont contp, TSEvent event, Data *const data)
{
  switch (event) {
  case TS_EVENT_VCONN_READ_READY:
  case TS_EVENT_VCONN_READ_COMPLETE: {
    if (nullptr == data->m_http_parser) {
      data->m_http_parser = TSHttpParserCreate();
    }

    // Read the header from the buffer
    int64_t consumed = 0;
    if (TS_PARSE_DONE !=
        data->m_req_hdrmgr.populateFrom(data->m_http_parser, data->m_dnstream.m_read.m_reader, TSHttpHdrParseReq, &consumed)) {
      return false;
    }

    // update the VIO
    TSVIO const input_vio = data->m_dnstream.m_read.m_vio;
    TSVIONDoneSet(input_vio, TSVIONDoneGet(input_vio) + consumed);

    // make the header manipulator
    HttpHeader header(data->m_req_hdrmgr.m_buffer, data->m_req_hdrmgr.m_lochdr);

    // set the request url back to pristine in case of plugin stacking
    header.setUrl(data->m_urlbuf, data->m_urlloc);

    header.setKeyVal(TS_MIME_FIELD_HOST, TS_MIME_LEN_HOST, data->m_hostname, data->m_hostlen);

    // Slice never sends a request body on the internal connection. Drop any
    // body-declaring fields so the internal transaction does not wait for a
    // body that will never arrive.
    header.removeKey(TS_MIME_FIELD_CONTENT_LENGTH, TS_MIME_LEN_CONTENT_LENGTH);
    header.removeKey(TS_MIME_FIELD_TRANSFER_ENCODING, TS_MIME_LEN_TRANSFER_ENCODING);

    // default: whole file (unknown, wait for first server response)
    Range rangebe;

    char                rangestr[1024];
    int                 rangelen = sizeof(rangestr);
    bool const          hasRange = header.valueForKey(TS_MIME_FIELD_RANGE, TS_MIME_LEN_RANGE, rangestr, &rangelen,
                                                      0); // <-- first range only
    Config const *const conf     = data->m_config;
    if (hasRange) {
      // write parsed header into slicer meta tag
      header.setKeyVal(conf->m_skip_header.c_str(), conf->m_skip_header.size(), rangestr, rangelen);
      bool const isRangeGood = rangebe.fromStringClosed(rangestr);

      if (isRangeGood) {
        DEBUG_LOG("%p Partial content request", data);
        data->m_statustype = TS_HTTP_STATUS_PARTIAL_CONTENT;
      } else // signal a 416 needs to be formed and sent
      {
        DEBUG_LOG("%p Ill formed/unhandled range: %s", data, rangestr);
        data->m_statustype = TS_HTTP_STATUS_REQUESTED_RANGE_NOT_SATISFIABLE;

        // First block will give Content-Length
        rangebe = Range(0, conf->m_blockbytes);
      }
    } else {
      DEBUG_LOG("%p Full content request", data);
      static char const *const valstr = "-";
      static size_t const      vallen = strlen(valstr);
      header.setKeyVal(conf->m_skip_header.data(), conf->m_skip_header.size(), valstr, vallen);
      data->m_statustype = TS_HTTP_STATUS_OK;
      rangebe            = Range(0, Range::maxval);
    }

    if (Config::RefType::First == conf->m_reftype) {
      data->m_blocknum = 0;
    } else {
      data->m_blocknum = rangebe.firstBlockFor(conf->m_blockbytes);
    }

    data->m_req_range = rangebe;

    if (data->is_purge()) {
      // The substituted range covers block 0, so walking it would delete the head
      if (TS_HTTP_STATUS_REQUESTED_RANGE_NOT_SATISFIABLE == data->m_statustype) {
        // paced: the range is client supplied, so a bad one repeats as fast as requests arrive
        if (data->m_config->canLogError()) {
          ERROR_LOG("Refusing PURGE with an unparseable range");
        }
        finish_purge(contp, data, TS_HTTP_STATUS_BAD_REQUEST);
        return true;
      }

      data->m_purge_miss_bound = purge_miss_bound(header, data->m_config);
      DEBUG_LOG("%p Purge miss bound %d block(s)", data, data->m_purge_miss_bound);

      // A suffix range cannot know its start block, so purge a superset: everything
      if (data->m_req_range.isEndBytes()) {
        data->m_req_range = Range(0, Range::maxval);
        data->m_blocknum  = 0;
        DEBUG_LOG("%p Purge suffix range widened to the whole object", data);
      }

      // The miss bound is for this proxy to act on, not to propagate
      header.removeKey(conf->m_purge_probe_header.data(), conf->m_purge_probe_header.size());
    }

    // remove ATS keys to avoid 404 loop
    header.removeKey(TS_MIME_FIELD_VIA, TS_MIME_LEN_VIA);
    header.removeKey(TS_MIME_FIELD_X_FORWARDED_FOR, TS_MIME_LEN_X_FORWARDED_FOR);

    // send block request to server
    if (!request_block(contp, data)) {
      abort(contp, data);
      return false;
    }

    // for subsequent blocks remove any conditionals which may fail
    // an optimization would be to wait until the first block succeeds
    header.removeKey(TS_MIME_FIELD_IF_MATCH, TS_MIME_LEN_IF_MATCH);
    header.removeKey(TS_MIME_FIELD_IF_MODIFIED_SINCE, TS_MIME_LEN_IF_MODIFIED_SINCE);
    header.removeKey(TS_MIME_FIELD_IF_NONE_MATCH, TS_MIME_LEN_IF_NONE_MATCH);
    header.removeKey(TS_MIME_FIELD_IF_RANGE, TS_MIME_LEN_IF_RANGE);
    header.removeKey(TS_MIME_FIELD_IF_UNMODIFIED_SINCE, TS_MIME_LEN_IF_UNMODIFIED_SINCE);
  } break;
  default: {
    DEBUG_LOG("%p handle_client_req unhandled event %d %s", data, event, TSHttpEventNameLookup(event));
  } break;
  }

  return true;
}

// this is when the client starts asking us for more data
void
handle_client_resp(TSCont contp, TSEvent event, Data *const data)
{
  switch (event) {
  case TS_EVENT_VCONN_WRITE_READY: {
    // finish_purge writes the whole response at once; nothing to throttle or pull
    if (data->is_purge()) {
      break;
    }

    switch (data->m_blockstate) {
    case BlockState::Fail:
    case BlockState::PendingRef:
    case BlockState::ActiveRef: {
      TSVIO const   output_vio  = data->m_dnstream.m_write.m_vio;
      int64_t const output_done = TSVIONDoneGet(output_vio);
      int64_t const output_sent = data->m_bytessent;

      if (output_sent == output_done) {
        DEBUG_LOG("Downstream output is done, shutting down");
        shutdown(contp, data);
      }
    } break;

    case BlockState::Pending: {
      // throttle
      TSVIO const   output_vio  = data->m_dnstream.m_write.m_vio;
      int64_t const output_done = TSVIONDoneGet(output_vio);
      int64_t const output_sent = data->m_bytessent;
      int64_t const threshout   = data->m_config->m_blockbytes;
      int64_t const buffered    = output_sent - output_done;

      if (threshout < buffered) {
        DEBUG_LOG("%p handle_client_resp: throttling %" PRId64, data, buffered);
      } else {
        DEBUG_LOG("Starting next block request");
        if (!request_block(contp, data)) {
          data->m_blockstate = BlockState::Fail;
          return;
        }
      }
    } break;
    case BlockState::Passthru: {
    } break;
    default:
      break;
    }
  } break;
  case TS_EVENT_VCONN_WRITE_COMPLETE: {
    if (dbg_ctl.on() && reader_avail_more_than(data->m_upstream.m_read.m_reader, 0)) {
      int64_t const left = TSIOBufferReaderAvail(data->m_upstream.m_read.m_reader);
      DEBUG_LOG("%p WRITE_COMPLETE called with %" PRId64 " bytes left", data, left);
    }

    data->m_dnstream.close();
    if (!data->m_upstream.m_read.isOpen()) {
      shutdown(contp, data);
    }
  } break;
  default: {
    DEBUG_LOG("%p handle_client_resp unhandled event %d %s", data, event, TSHttpEventNameLookup(event));
  } break;
  }
}