text: #include "extensions/filters/network/mysql_proxy/mysql_filter.h"
  
  #include "envoy/config/core/v3alpha/base.pb.h"
  
  #include "common/buffer/buffer_impl.h"
  #include "common/common/assert.h"
  #include "common/common/logger.h"
  
  #include "extensions/filters/network/well_known_names.h"
  
  #include "include/sqlparser/SQLParser.h"
  
  namespace Envoy {
  namespace Extensions {
  namespace NetworkFilters {
  namespace MySQLProxy {
  
  MySQLFilterConfig::MySQLFilterConfig(const std::string& stat_prefix, Stats::Scope& scope)
      : scope_(scope), stat_prefix_(stat_prefix), stats_(generateStats(stat_prefix, scope)) {}
  
  MySQLFilter::MySQLFilter(MySQLFilterConfigSharedPtr config) : config_(std::move(config)) {}
  
  void MySQLFilter::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {
    read_callbacks_ = &callbacks;
  }
  
  Network::FilterStatus MySQLFilter::onData(Buffer::Instance& data, bool) {
    // Safety measure just to make sure that if we have a decoding error we keep going and lose stats.
    // This can be removed once we are more confident of this code.
    if (sniffing_) {
      read_buffer_.add(data);
      doDecode(read_buffer_);
    }
    return Network::FilterStatus::Continue;
  }
  
  Network::FilterStatus MySQLFilter::onWrite(Buffer::Instance& data, bool) {
    // Safety measure just to make sure that if we have a decoding error we keep going and lose stats.
    // This can be removed once we are more confident of this code.
    if (sniffing_) {
      write_buffer_.add(data);
      doDecode(write_buffer_);
    }
    return Network::FilterStatus::Continue;
  }
  
  void MySQLFilter::doDecode(Buffer::Instance& buffer) {
    // Clear dynamic metadata.
    envoy::config::core::v3alpha::Metadata& dynamic_metadata =
        read_callbacks_->connection().streamInfo().dynamicMetadata();
    auto& metadata =
        (*dynamic_metadata.mutable_filter_metadata())[NetworkFilterNames::get().MySQLProxy];
    metadata.mutable_fields()->clear();
  
    if (!decoder_) {
      decoder_ = createDecoder(*this);
    }
  
    try {
      decoder_->onData(buffer);
    } catch (EnvoyException& e) {
      ENVOY_LOG(info, "mysql_proxy: decoding error: {}", e.what());
      config_->stats_.decoder_errors_.inc();
      sniffing_ = false;
      read_buffer_.drain(read_buffer_.length());
      write_buffer_.drain(write_buffer_.length());
    }
  }
  
  DecoderPtr MySQLFilter::createDecoder(DecoderCallbacks& callbacks) {
    return std::make_unique<DecoderImpl>(callbacks);
  }
  
  void MySQLFilter::onProtocolError() { config_->stats_.protocol_errors_.inc(); }
  
  void MySQLFilter::onNewMessage(MySQLSession::State state) {
    if (state == MySQLSession::State::ChallengeReq) {
      config_->stats_.login_attempts_.inc();
    }
  }
  
  void MySQLFilter::onClientLogin(ClientLogin& client_login) {
    if (client_login.isSSLRequest()) {
      config_->stats_.upgraded_to_ssl_.inc();
    }
  }
  
  void MySQLFilter::onClientLoginResponse(ClientLoginResponse& client_login_resp) {
    if (client_login_resp.getRespCode() == MYSQL_RESP_AUTH_SWITCH) {
      config_->stats_.auth_switch_request_.inc();
    } else if (client_login_resp.getRespCode() == MYSQL_RESP_ERR) {
      config_->stats_.login_failures_.inc();
    }
  }
  
  void MySQLFilter::onMoreClientLoginResponse(ClientLoginResponse& client_login_resp) {
    if (client_login_resp.getRespCode() == MYSQL_RESP_ERR) {
      config_->stats_.login_failures_.inc();
    }
  }
  
  void MySQLFilter::onCommand(Command& command) {
    if (!command.isQuery()) {
      return;
    }
  
    // Parse a given query
    hsql::SQLParserResult result;
    hsql::SQLParser::parse(command.getData(), &result);
  
    ENVOY_CONN_LOG(trace, "mysql_proxy: query processed {}", read_callbacks_->connection(),
                   command.getData());
  
    if (!result.isValid()) {
      config_->stats_.queries_parse_error_.inc();
      return;
    }
    config_->stats_.queries_parsed_.inc();
  
    // Set dynamic metadata
    envoy::config::core::v3alpha::Metadata& dynamic_metadata =
        read_callbacks_->connection().streamInfo().dynamicMetadata();
    ProtobufWkt::Struct metadata(
        (*dynamic_metadata.mutable_filter_metadata())[NetworkFilterNames::get().MySQLProxy]);
    auto& fields = *metadata.mutable_fields();
  
    for (auto i = 0u; i < result.size(); ++i) {
      if (result.getStatement(i)->type() == hsql::StatementType::kStmtShow) {
        continue;
      }
      hsql::TableAccessMap table_access_map;
      result.getStatement(i)->tablesAccessed(table_access_map);
      for (auto& it : table_access_map) {
        auto& operations = *fields[it.first].mutable_list_value();
        for (const auto& ot : it.second) {
          operations.add_values()->set_string_value(ot);
        }
      }
    }
  
    read_callbacks_->connection().streamInfo().setDynamicMetadata(
        NetworkFilterNames::get().MySQLProxy, metadata);
  }
  
  Network::FilterStatus MySQLFilter::onNewConnection() {
    config_->stats_.sessions_.inc();
    return Network::FilterStatus::Continue;
  }
  
  } // namespace MySQLProxy
  } // namespace NetworkFilters
  } // namespace Extensions
  } // namespace Envoy
  