diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5f2f7b6ce3..bae1620b1b 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -570,8 +570,6 @@ add_library (seastar
   include/seastar/core/rwlock.hh
   include/seastar/core/scattered_message.hh
   include/seastar/core/scheduling.hh
-  include/seastar/core/scollectd.hh
-  include/seastar/core/scollectd_api.hh
   include/seastar/core/seastar.hh
   include/seastar/core/semaphore.hh
   include/seastar/core/shard_id.hh
@@ -714,8 +712,6 @@ add_library (seastar
   src/core/reactor.cc
   src/core/resource.cc
   src/core/sharded.cc
-  src/core/scollectd.cc
-  src/core/scollectd-impl.hh
   src/core/signal.cc
   src/core/systemwide_memory_barrier.cc
   src/core/smp.cc
diff --git a/include/seastar/core/app-template.hh b/include/seastar/core/app-template.hh
index e807a2f9ba..4378185756 100644
--- a/include/seastar/core/app-template.hh
+++ b/include/seastar/core/app-template.hh
@@ -31,7 +31,6 @@
 #include <seastar/core/sstring.hh>
 #include <seastar/util/program-options.hh>
 #include <seastar/core/metrics_api.hh>
-#include <seastar/core/scollectd.hh>
 #include <seastar/util/log-cli.hh>
 #include <seastar/util/modules.hh>
 
@@ -112,8 +111,6 @@ public:
         metrics::options metrics_opts;
         /// Configuration options for the smp aspect of seastar.
         smp_options smp_opts;
-        /// Configuration for the scollectd sub-system.
-        scollectd::options scollectd_opts;
         /// Configuration for the logging sub-system.
         log_cli::options log_opts;
 
diff --git a/include/seastar/core/scollectd.hh b/include/seastar/core/scollectd.hh
deleted file mode 100644
index 732e054fcd..0000000000
--- a/include/seastar/core/scollectd.hh
+++ /dev/null
@@ -1,866 +0,0 @@
-/*
- * This file is open source software, licensed to you under the terms
- * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
- * distributed with this work for additional information regarding copyright
- * ownership.  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.
- */
-/*
- * Copyright (C) 2014 Cloudius Systems, Ltd.
- */
-
-#pragma once
-
-#ifndef SEASTAR_MODULE
-#include <type_traits>
-#include <utility>
-#include <functional>
-#include <array>
-#include <iterator>
-#include <stdint.h>
-#include <memory>
-#include <string>
-#include <tuple>
-#include <chrono>
-#endif
-
-#include <seastar/core/future.hh>
-#include <seastar/net/byteorder.hh>
-#include <seastar/core/shared_ptr.hh>
-#include <seastar/core/sstring.hh>
-#include <seastar/util/log.hh>
-#include <seastar/util/program-options.hh>
-#include <seastar/util/modules.hh>
-#include <seastar/core/metrics_api.hh>
-
-namespace seastar {
-
-/**
- * Implementation of rudimentary collectd data gathering.
- *
- * Usage is hopefully straight forward. Though, feel free to read
- * https://collectd.org/wiki/index.php/Naming_schema
- * for an explanation on the naming model.
- *
- * Typically, you'll add values something like:
- *
- * 		scollectd::type_instance_id typ("<pluginname>", "<instance_name>", "<type_name>", "<instance_name>");
- *		scollectd::add_polled_metric(typ, [<metric var> | scollectd::make_typed(<data_type>, <metric_var>) [, ...]);
- *
- * Where
- * 	`<pluginname>` would be the overall 'module', e.g. "cpu"
- *  `<instance_name>` -> optional distinguisher between plugin instances. For cpu, the built-in
- *  scollectd::per_cpu_plugin_instance constant is a good choice, i.e. 0->N cpu.
- *  If there are no instances (e.g. only one), empty constant is appropriate (none)
- *  `<type_name>` is the 'type' of metric collected, for ex. "usage" (cpu/0/usage)
- *  `<type_instance>` is a distinguisher for metric parts of the type, e.g. "idle", "user", "kernel"
- *  -> cpu/0/usage/idle | cpu/0/usage/user | cpu/0/usage/kernel
- *
- *  Each type instance can bind an arbitrary number of values, ech representing some aspect in turn of the instance.
- *  The structure and interpretation is up to the producer/consumer
- *
- * There is a single "scollectd" instance per cpu, and values should be bound locally
- * to this cpu. Polling is done at a frequency set in the seastar config (def once per s),
- * and all registered values will be sent via UDP packages to the destination host(s)
- *
- * Note that the tuple { plugin, plugin_instance, type, type_instance } is considered a
- * unique ID for a value registration, so using the same tuple twice will remove the previously
- * registered values.
- *
- * Values can be unregistered at any time, though they must be so on the same thread/cpu
- * as they we're registered. The "registration" achor type provides RAII style value unregistration
- * semantics.
- *
- */
-
-namespace scollectd {
-
-extern seastar::logger logger;
-
-using data_type = seastar::metrics::impl::data_type;
-
-enum class known_type {
-    // from types.db. Defined collectd types (type_id) selection.
-    // This enum omits the very application specific types, such
-    // as mysql_* etc, since if you really are re-writing mysql
-    // in seastar, you probably know how to look the type up manually...
-
-    absolute,
-    backends,
-    bitrate,
-    blocked_clients,
-    bytes,
-    cache_eviction,
-    cache_operation,
-    cache_ratio,
-    cache_result,
-    cache_size,
-    capacity,
-    changes_since_last_save,
-    charge,
-    clock_last_meas,
-    clock_last_update,
-    clock_mode,
-    clock_reachability,
-    clock_skew_ppm,
-    clock_state,
-    clock_stratum,
-    compression,
-    compression_ratio,
-    connections,
-    conntrack,
-    contextswitch,
-    count,
-    counter,
-    cpu,
-    cpufreq,
-    current,
-    current_connections,
-    current_sessions,
-    delay,
-    derive,
-    df,
-    df_complex,
-    df_inodes,
-    disk_io_time,
-    disk_latency,
-    disk_merged,
-    disk_octets,
-    disk_ops,
-    disk_ops_complex,
-    disk_time,
-    dns_answer,
-    dns_notify,
-    dns_octets,
-    dns_opcode,
-    dns_qtype,
-    dns_qtype_cached,
-    dns_query,
-    dns_question,
-    dns_rcode,
-    dns_reject,
-    dns_request,
-    dns_resolver,
-    dns_response,
-    dns_transfer,
-    dns_update,
-    dns_zops,
-    drbd_resource,
-    duration,
-    email_check,
-    email_count,
-    email_size,
-    entropy,
-    evicted_keys,
-    expired_keys,
-    fanspeed,
-    file_handles,
-    file_size,
-    files,
-    flow,
-    fork_rate,
-    frequency,
-    frequency_error,
-    frequency_offset,
-    fscache_stat,
-    gauge,
-    hash_collisions,
-    http_request_methods,
-    http_requests,
-    http_response_codes,
-    humidity,
-    if_collisions,
-    if_dropped,
-    if_errors,
-    if_multicast,
-    if_octets,
-    if_packets,
-    if_rx_errors,
-    if_rx_octets,
-    if_tx_errors,
-    if_tx_octets,
-    invocations,
-    io_octets,
-    io_packets,
-    ipt_bytes,
-    ipt_packets,
-    irq,
-    latency,
-    links,
-    load,
-    md_disks,
-    memory,
-    memory_lua,
-    memory_throttle_count,
-    multimeter,
-    mutex_operations,
-    objects,
-    operations,
-    packets,
-    pending_operations,
-    percent,
-    percent_bytes,
-    percent_inodes,
-    ping,
-    ping_droprate,
-    ping_stddev,
-    players,
-    power,
-    pressure,
-    protocol_counter,
-    pubsub,
-    queue_length,
-    records,
-    requests,
-    response_code,
-    response_time,
-    root_delay,
-    root_dispersion,
-    route_etx,
-    route_metric,
-    routes,
-    segments,
-    serial_octets,
-    signal_noise,
-    signal_power,
-    signal_quality,
-    snr,
-    spl,
-    swap,
-    swap_io,
-    tcp_connections,
-    temperature,
-    threads,
-    time_dispersion,
-    time_offset,
-    time_offset_ntp,
-    time_offset_rms,
-    time_ref,
-    timeleft,
-    total_bytes,
-    total_connections,
-    total_objects,
-    total_operations,
-    total_requests,
-    total_sessions,
-    total_threads,
-    total_time_in_ms,
-    total_values,
-    uptime,
-    users,
-    vcl,
-    vcpu,
-    virt_cpu_total,
-    virt_vcpu,
-    vmpage_action,
-    vmpage_faults,
-    vmpage_io,
-    vmpage_number,
-    volatile_changes,
-    voltage,
-    voltage_threshold,
-    vs_memory,
-    vs_processes,
-    vs_threads,
-};
-
-// don't use directly. use make_typed.
-template<typename T>
-struct typed {
-    typed(data_type t, T && v)
-    : type(t), value(std::forward<T>(v)) {
-    }
-    data_type type;
-    T value;
-};
-
-template<typename T>
-inline typed<T> make_typed(data_type type, T&& t) {
-    return typed<T>(type, std::forward<T>(t));
-}
-
-using plugin_id = seastar::metrics::group_name_type;
-using plugin_instance_id = seastar::metrics::instance_id_type;
-using type_id = seastar::metrics::metric_type_def;
-using type_instance = seastar::metrics::metric_name_type;
-
-type_id type_id_for(known_type);
-
-using description = seastar::metrics::description;
-
-static constexpr unsigned max_collectd_field_text_len = 63;
-
-class type_instance_id {
-    static thread_local unsigned _next_truncated_idx;
-
-    /// truncate a given field to the maximum allowed length
-    void truncate(sstring& field, const char* field_desc);
-public:
-    type_instance_id() = default;
-    type_instance_id(plugin_id p, plugin_instance_id pi, type_id t,
-                    scollectd::type_instance ti = std::string())
-                    : _plugin(std::move(p)), _plugin_instance(std::move(pi)), _type(
-                                    std::move(t)), _type_instance(std::move(ti)) {
-        // truncate strings to the maximum allowed length
-        truncate(_plugin, "plugin");
-        truncate(_plugin_instance, "plugin_instance");
-        truncate(_type, "type");
-        truncate(_type_instance, "type_instance");
-    }
-    type_instance_id(const seastar::metrics::impl::metric_id &id, const type_id& inherit_type) : _plugin(id.group_name()),
-            _plugin_instance(id.instance_id()), _type(inherit_type),
-            _type_instance(id.name()) {
-    }
-    type_instance_id(type_instance_id &&) = default;
-    type_instance_id(const type_instance_id &) = default;
-
-    type_instance_id & operator=(type_instance_id &&) = default;
-    type_instance_id & operator=(const type_instance_id &) = default;
-
-    const plugin_id & plugin() const {
-        return _plugin;
-    }
-    const plugin_instance_id & plugin_instance() const {
-        return _plugin_instance;
-    }
-    const type_id & type() const {
-        return _type;
-    }
-    const scollectd::type_instance & type_instance() const {
-        return _type_instance;
-    }
-    bool operator<(const type_instance_id&) const;
-    bool operator==(const type_instance_id&) const;
-private:
-    plugin_id _plugin;
-    plugin_instance_id _plugin_instance;
-    type_id _type;
-    scollectd::type_instance _type_instance;
-};
-
-extern const plugin_instance_id per_cpu_plugin_instance;
-
-// Scollectd configuration options.
-struct options : public program_options::option_group {
-    /// \brief Enable collectd daemon.
-    ///
-    /// Default: \p false.
-    program_options::value<bool> collectd;
-    /// \brief Address to send/broadcast metrics to.
-    ///
-    /// Default: \p 239.192.74.66:25826.
-    program_options::value<std::string> collectd_address;
-    /// \brief Poll period (ms).
-    ///
-    /// Frequency of sending counter metrics (0 disables).
-    /// Default: \p 1000.
-    program_options::value<unsigned> collectd_poll_period;
-    /// \deprecated use \ref metrics::options::metrics_hostname instead
-    program_options::value<std::string> collectd_hostname;
-
-    /// \cond internal
-    options(program_options::option_group* parent_group);
-    /// \endcond
-};
-
-void configure(const options&);
-void remove_polled_metric(const type_instance_id &);
-
-class plugin_instance_metrics;
-
-/**
- * Anchor for polled registration.
- * Iff the registered type is in some way none-persistent,
- * use this as receiver of the reg and ensure it dies before the
- * added value(s).
- *
- * Use:
- * uint64_t v = 0;
- * registration r = add_polled_metric(v);
- * ++r;
- * <scope end, above dies>
- */
-struct registration {
-    registration() = default;
-    registration(const type_instance_id& id);
-    registration(type_instance_id&& id);
-    registration(const registration&) = delete;
-    registration(registration&&) = default;
-    ~registration();
-    registration & operator=(const registration&) = delete;
-    registration & operator=(registration&&) = default;
-
-    void unregister() {
-        remove_polled_metric(_id);
-        _id = type_instance_id();
-    }
-private:
-    friend class plugin_instance_metrics;
-    type_instance_id _id;
-    shared_ptr<seastar::metrics::impl::impl> _impl;
-};
-
-/**
- * Helper type to make generating vectors of registration objects
- * easier, since it constructs from an initializer list of
- * type_instance_id:s, avoiding early conversion to registration objs,
- * which in case of init lists, are copy semantics, not move...
- */
-class registrations
-    : public std::vector<registration>
-{
-public:
-    typedef std::vector<registration> vector_type;
-
-    registrations()
-    {}
-    registrations(vector_type&& v) : vector_type(std::move(v))
-    {}
-    registrations(const std::initializer_list<type_instance_id>& l)
-        : vector_type(l.begin(),l.end())
-    {}
-    registrations& operator=(vector_type&& v) {
-        vector_type::operator=(std::move(v));
-        return *this;
-    }
-    registrations& operator=(const std::initializer_list<type_instance_id>& l) {
-        return registrations::operator=(registrations(l));
-    }
-};
-
-class value_list;
-
-struct typed_value {
-    /**
-     * Wraps N values of a given type (type_id).
-     * Used to group types into a plugin_instance_metrics
-     */
-    template<typename... Args>
-    typed_value(const type_id& tid, const scollectd::type_instance& ti, description, Args&&... args);
-
-    template<typename... Args>
-    typed_value(const type_id& tid, const scollectd::type_instance& ti, Args&&... args)
-        : typed_value(tid, ti, description(), std::forward<Args>(args)...)
-    {}
-
-    const scollectd::type_instance& type_instance() const {
-        return _type_instance;
-    }
-    const shared_ptr<value_list>& values() const {
-        return _values;
-    }
-    const type_id & type() const {
-        return _type_id;
-    }
-private:
-    type_id _type_id;
-    scollectd::type_instance _type_instance;
-    shared_ptr<value_list> _values;
-};
-
-class plugin_instance_metrics {
-public:
-    template<typename... TypedValues>
-    plugin_instance_metrics(const plugin_id& p, const plugin_instance_id& pi, TypedValues&&... values)
-        : _plugin_id(p)
-        , _plugin_instance(pi)
-        , _registrations({ add_impl(values)... })
-    {}
-    std::vector<type_instance_id> bound_ids() const;
-    void add(const typed_value&);
-private:
-    type_instance_id add_impl(const typed_value&);
-
-    plugin_id _plugin_id;
-    plugin_instance_id _plugin_instance;
-    registrations _registrations;
-};
-
-/**
- * Simplified wrapper for the common case of per-cpu plugin instances
- * (i.e. distributed objects)
- */
-class percpu_plugin_instance_metrics : public plugin_instance_metrics {
-public:
-    template<typename... TypedValues>
-    percpu_plugin_instance_metrics(const plugin_id& p, TypedValues&&... values)
-        : plugin_instance_metrics(p, per_cpu_plugin_instance, std::forward<TypedValues>(values)...)
-    {}
-};
-
-/**
- * Template wrapper for type_id values, deriving type_id string
- * from the known_types enum, for auto-completetion joy.
- */
-template<known_type Type>
-struct typed_value_impl: public typed_value {
-    template<typename ... Args>
-    typed_value_impl(const scollectd::type_instance& ti, Args&& ... args)
-        : typed_value(type_id_for(Type), ti, std::forward<Args>(args)...)
-    {}
-
-    template<typename ... Args>
-    typed_value_impl(scollectd::type_instance ti, description d, Args&& ... args)
-        : typed_value(type_id_for(Type), std::move(ti), std::move(d), std::forward<Args>(args)...)
-    {}
-    template<typename ... Args>
-    typed_value_impl(description d, Args&& ... args)
-        : typed_value(type_id_for(Type), scollectd::type_instance(), std::move(d), std::forward<Args>(args)...)
-    {}
-};
-
-/*!
- * \deprecated metrics registration should be done using the metrics layer
- *
- * Some typedefs for common used types. Feel free to add.
- */
-typedef typed_value_impl<known_type::total_bytes> total_bytes;
-typedef typed_value_impl<known_type::total_connections> total_connections;
-typedef typed_value_impl<known_type::total_objects> total_objects;
-typedef typed_value_impl<known_type::total_operations> total_operations;
-typedef typed_value_impl<known_type::total_requests> total_requests;
-typedef typed_value_impl<known_type::total_sessions> total_sessions;
-typedef typed_value_impl<known_type::total_threads> total_threads;
-typedef typed_value_impl<known_type::total_time_in_ms> total_time_in_ms;
-typedef typed_value_impl<known_type::total_values> total_values;
-typedef typed_value_impl<known_type::queue_length> queue_length;
-typedef typed_value_impl<known_type::counter> counter;
-typedef typed_value_impl<known_type::count> count;
-typedef typed_value_impl<known_type::gauge> gauge;
-
-// lots of template junk to build typed value list tuples
-// for registered values.
-template<typename T, typename En = void>
-struct data_type_for;
-
-template<typename T, typename En = void>
-struct is_callable;
-
-template<typename T>
-struct is_callable<T,
-std::enable_if_t<
-!std::is_void_v<std::invoke_result_t<T>>,
-void>> : public std::true_type {
-};
-
-template<typename T>
-struct is_callable<T,
-std::enable_if_t<std::is_fundamental_v<T>, void>> : public std::false_type {
-};
-
-template<typename T>
-struct data_type_for<T,
-std::enable_if_t<
-std::is_integral_v<T> && std::is_unsigned_v<T>,
-void>> : public std::integral_constant<data_type,
-data_type::COUNTER> {
-};
-template<typename T>
-struct data_type_for<T,
-std::enable_if_t<std::is_floating_point_v<T>, void>> : public std::integral_constant<
-data_type, data_type::GAUGE> {
-};
-template<typename T>
-struct data_type_for<T,
-std::enable_if_t<is_callable<T>::value, void>> : public data_type_for<
-std::invoke_result_t<T>> {
-};
-template<typename T>
-struct data_type_for<typed<T>> : public data_type_for<T> {
-};
-
-template<typename T>
-class value {
-public:
-    template<typename W>
-    struct wrap {
-        wrap(const W & v)
-        : _v(v) {
-        }
-        const W & operator()() const {
-            return _v;
-        }
-        const W & _v;
-    };
-
-    typedef std::remove_reference_t<T> value_type;
-    typedef std::conditional_t<
-            is_callable<std::remove_reference_t<T>>::value,
-            value_type, wrap<value_type> > stored_type;
-
-    value(const value_type & t)
-    : value<T>(data_type_for<value_type>::value, t) {
-    }
-    value(data_type type, const value_type & t)
-    : _type(type), _t(t) {
-    }
-    uint64_t operator()() const {
-        auto v = _t();
-        if (_type == data_type::GAUGE) {
-            return convert(double(v));
-        } else {
-            uint64_t u = v;
-            return convert(u);
-        }
-    }
-    operator uint64_t() const {
-        return (*this)();
-    }
-    operator data_type() const {
-        return _type;
-    }
-    data_type type() const {
-        return _type;
-    }
-private:
-    // not super quick value -> protocol endian 64-bit values.
-    template<typename _Iter>
-    void bpack(_Iter s, _Iter e, uint64_t v) const {
-        while (s != e) {
-            *s++ = (v & 0xff);
-            v >>= 8;
-        }
-    }
-    template<typename V>
-    std::enable_if_t<std::is_integral_v<V>, uint64_t> convert(
-            V v) const {
-        uint64_t i = v;
-        // network byte order
-        return ntohq(i);
-    }
-    template<typename V>
-    std::enable_if_t<std::is_floating_point_v<V>, uint64_t> convert(
-            V t) const {
-        union {
-            uint64_t i;
-            double v;
-        } v;
-        union {
-            uint64_t i;
-            uint8_t b[8];
-        } u;
-        v.v = t;
-        // intel byte order. could also obviously be faster.
-        // could be ignored if we just assume we're le (for now),
-        // but this is ok me thinks.
-        bpack(std::begin(u.b), std::end(u.b), v.i);
-        return u.i;
-    }
-    ;
-
-    const data_type _type;
-    const stored_type _t;
-};
-
-template<typename T>
-class value<typed<T>> : public value<T> {
-public:
-    value(const typed<T> & args)
-: value<T>(args.type, args.value) {
-    }
-};
-
-class value_list {
-    bool _enabled = true;
-public:
-    value_list(description d) : _description(std::move(d))
-    {}
-    value_list(value_list&&) = default;
-    virtual ~value_list() {}
-
-    virtual size_t size() const = 0;
-
-    virtual void types(data_type *) const = 0;
-    virtual void values(net::packed<uint64_t> *) const = 0;
-
-    const description& desc() const {
-        return _description;
-    }
-
-    bool empty() const {
-        return size() == 0;
-    }
-
-    bool is_enabled() const {
-        return _enabled;
-    }
-
-    void set_enabled(bool b) {
-        _enabled = b;
-    }
-private:
-    description _description;
-};
-
-template<typename ... Args>
-class values_impl: public value_list {
-public:
-    static const size_t num_values = sizeof...(Args);
-
-    values_impl(description d, Args&& ...args)
-        : value_list(std::move(d))
-        , _values(std::forward<Args>(args)...)
-    {}
-
-    values_impl(values_impl<Args...>&& a) = default;
-    values_impl(const values_impl<Args...>& a) = default;
-
-    size_t size() const override {
-        return num_values;
-    }
-    void types(data_type * p) const override {
-        unpack(_values, [p](Args... args) {
-            std::initializer_list<data_type> tmp = { args...  };
-            std::copy(tmp.begin(), tmp.end(), p);
-        });
-    }
-    void values(net::packed<uint64_t> * p) const override {
-        unpack(_values, [p](Args... args) {
-            std::initializer_list<uint64_t> tmp = { args...  };
-            std::copy(tmp.begin(), tmp.end(), p);
-        });
-    }
-private:
-    template<typename _Op>
-    void unpack(const std::tuple<Args...>& t, _Op&& op) const {
-        do_unpack(t, std::index_sequence_for<Args...> {}, std::forward<_Op>(op));
-    }
-
-    template<size_t ...S, typename _Op>
-    void do_unpack(const std::tuple<Args...>& t, const std::index_sequence<S...> &, _Op&& op) const {
-        op(std::get<S>(t)...);
-    }
-
-    std::tuple < Args... > _values;
-};
-
-void add_polled(const type_instance_id &, const shared_ptr<value_list> &, bool enabled = true);
-
-typedef std::function<void()> notify_function;
-template<typename... _Args>
-static auto make_type_instance(description d, _Args && ... args) -> values_impl < decltype(value<_Args>(std::forward<_Args>(args)))... >
-{
-    return values_impl<decltype(value<_Args>(std::forward<_Args>(args)))...>(
-                    std::move(d), value<_Args>(std::forward<_Args>(args))...);
-}
-/*!
- * \deprecated metrics registration should be done using the metrics layer
- *
- */
-template<typename ... _Args>
-[[deprecated("Use the metrics layer")]] static type_instance_id add_polled_metric(const plugin_id & plugin,
-        const plugin_instance_id & plugin_instance, const type_id & type,
-        const scollectd::type_instance & type_instance, _Args&& ... args) {
-    return add_polled_metric(plugin, plugin_instance, type, type_instance, description(),
-            std::forward<_Args>(args)...);
-}
-/*!
- * \deprecated metrics registration should be done using the metrics layer
- *
- */
-template<typename ... _Args>
-[[deprecated("Use the metrics layer")]] static type_instance_id add_polled_metric(const plugin_id & plugin,
-        const plugin_instance_id & plugin_instance, const type_id & type,
-        const scollectd::type_instance & type_instance, description d, _Args&& ... args) {
-    return add_polled_metric(
-            type_instance_id(plugin, plugin_instance, type, type_instance), std::move(d),
-            std::forward<_Args>(args)...);
-}
-template<typename ... _Args>
-static future<> send_explicit_metric(const plugin_id & plugin,
-        const plugin_instance_id & plugin_instance, const type_id & type,
-        const scollectd::type_instance & type_instance, _Args&& ... args) {
-    return send_explicit_metric(
-            type_instance_id(plugin, plugin_instance, type, type_instance),
-            std::forward<_Args>(args)...);
-}
-template<typename ... _Args>
-static notify_function create_explicit_metric(const plugin_id & plugin,
-        const plugin_instance_id & plugin_instance, const type_id & type,
-        const scollectd::type_instance & type_instance, _Args&& ... args) {
-    return create_explicit_metric(
-            type_instance_id(plugin, plugin_instance, type, type_instance),
-            std::forward<_Args>(args)...);
-}
-
-seastar::metrics::impl::metric_id to_metrics_id(const type_instance_id & id);
-/*!
- * \deprecated metrics registration should be done using the metrics layer
- *
- */
-template<typename Arg>
-[[deprecated("Use the metrics layer")]] static type_instance_id add_polled_metric(const type_instance_id & id, description d,
-        Arg&& arg, bool enabled = true) {
-    seastar::metrics::impl::get_local_impl()->add_registration(to_metrics_id(id), arg.type, seastar::metrics::impl::make_function(arg.value, arg.type), d, enabled);
-    return id;
-}
-/*!
- * \deprecated metrics registration should be done using the metrics layer
- *
- */
-template<typename Arg>
-[[deprecated("Use the metrics layer")]] static type_instance_id add_polled_metric(const type_instance_id & id,
-        Arg&& arg) {
-    return std::move(add_polled_metric(id, description(), std::forward<Arg>(arg)));
-}
-
-/*!
- * \deprecated metrics registration should be done using the metrics layer
- *
- */
-template<typename Args>
-[[deprecated("Use the metrics layer")]] static type_instance_id add_disabled_polled_metric(const type_instance_id & id, description d,
-        Args&& arg) {
-    return add_polled_metric(id, d, std::forward<Args>(arg), false);
-}
-
-template<typename Args>
-static type_instance_id add_disabled_polled_metric(const type_instance_id & id,
-        Args&& args) {
-    return add_disabled_polled_metric(id, description(), std::forward<Args>(args));
-}
-
-template<typename ... Args>
-static type_instance_id add_disabled_polled_metric(const type_instance_id & id,
-        Args&& ... args) {
-    return add_disabled_polled_metric(id, description(), std::forward<Args>(args)...);
-}
-
-// "Explicit" metric sends. Sends a single value list as a message.
-// Obviously not super efficient either. But maybe someone needs it sometime.
-template<typename ... _Args>
-static future<> send_explicit_metric(const type_instance_id & id,
-        _Args&& ... args) {
-    return send_metric(id, make_type_instance(std::forward<_Args>(args)...));
-}
-template<typename ... _Args>
-static notify_function create_explicit_metric(const type_instance_id & id,
-        _Args&& ... args) {
-    auto list = make_type_instance(std::forward<_Args>(args)...);
-    return [id, list=std::move(list)]() {
-        send_metric(id, list);
-    };
-}
-
-template<typename... Args>
-typed_value::typed_value(const type_id& tid, const scollectd::type_instance& ti, description d, Args&&... args)
-    : _type_id(tid)
-    , _type_instance(ti)
-    , _values(::seastar::make_shared<decltype(make_type_instance(std::move(d), std::forward<Args>(args)...))>(make_type_instance(std::move(d), std::forward<Args>(args)...)))
-{}
-
-// Send a message packet (string)
-future<> send_notification(const type_instance_id & id, const sstring & msg);
-}
-
-}
diff --git a/include/seastar/core/scollectd_api.hh b/include/seastar/core/scollectd_api.hh
deleted file mode 100644
index fe1bd0ed91..0000000000
--- a/include/seastar/core/scollectd_api.hh
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Copyright 2015 Cloudius Systems
- */
-
-#pragma once
-
-#ifndef SEASTAR_MODULE
-#include <seastar/core/scollectd.hh>
-#include <seastar/core/metrics_api.hh>
-#include <seastar/util/modules.hh>
-#include <vector>
-#endif
-
-namespace seastar {
-
-namespace scollectd {
-
-SEASTAR_MODULE_EXPORT_BEGIN
-
-using collectd_value  = seastar::metrics::impl::metric_value;
-
-std::vector<collectd_value> get_collectd_value(
-        const scollectd::type_instance_id& id);
-
-std::vector<scollectd::type_instance_id> get_collectd_ids();
-
-sstring get_collectd_description_str(const scollectd::type_instance_id&);
-
-bool is_enabled(const scollectd::type_instance_id& id);
-/**
- * Enable or disable collectd metrics on local instance
- * @param id - the metric to enable or disable
- * @param enable - should the collectd metrics be enable or disable
- */
-void enable(const scollectd::type_instance_id& id, bool enable);
-
-
-metrics::impl::value_map get_value_map();
-
-SEASTAR_MODULE_EXPORT_END
-
-}
-
-}
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 46e7a13daf..108d06ff1c 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -32,7 +32,6 @@ target_sources (seastar-module
     core/reactor_backend.cc
     core/resource.cc
     core/sharded.cc
-    core/scollectd.cc
     core/semaphore.cc
     core/signal.cc
     core/smp.cc
diff --git a/src/core/app-template.cc b/src/core/app-template.cc
index af76bda372..19f5393586 100644
--- a/src/core/app-template.cc
+++ b/src/core/app-template.cc
@@ -37,7 +37,6 @@ module seastar;
 #include <seastar/core/app-template.hh>
 #include <seastar/core/reactor.hh>
 #include <seastar/core/alien.hh>
-#include <seastar/core/scollectd.hh>
 #include <seastar/core/metrics_api.hh>
 #include <seastar/core/print.hh>
 #include <seastar/util/log.hh>
@@ -70,7 +69,6 @@ app_template::seastar_options::seastar_options()
     , reactor_opts(this)
     , metrics_opts(this)
     , smp_opts(this)
-    , scollectd_opts(this)
     , log_opts(this)
 {
 }
@@ -258,11 +256,7 @@ app_template::run_deprecated(int ac, char ** av, std::function<void ()>&& func)
     // No need to wait for this future.
     // func is waited on via engine().run()
     (void)engine().when_started().then([this] {
-        return seastar::metrics::configure(_opts.metrics_opts).then([this] {
-            // set scollectd use the metrics configuration, so the later
-            // need to be set first
-            scollectd::configure( _opts.scollectd_opts);
-        });
+        return seastar::metrics::configure(_opts.metrics_opts);
     }).then(
         std::move(func)
     ).then_wrapped([] (auto&& f) {
diff --git a/src/core/prometheus.cc b/src/core/prometheus.cc
index 06529cf79e..15626adaea 100644
--- a/src/core/prometheus.cc
+++ b/src/core/prometheus.cc
@@ -160,11 +160,11 @@ static void fill_native_type_histogram(const metrics::histogram& h, ::io::promet
 static void fill_metric(pm::MetricFamily& mf, const metrics::impl::metric_value& c,
         const metrics::impl::labels_type & id, const config& ctx) {
     switch (c.type()) {
-    case scollectd::data_type::GAUGE:
+    case seastar::metrics::impl::data_type::GAUGE:
         add_label(mf.add_metric(), id, ctx)->mutable_gauge()->set_value(c.d());
         mf.set_type(pm::MetricType::GAUGE);
         break;
-    case scollectd::data_type::SUMMARY: {
+    case seastar::metrics::impl::data_type::SUMMARY: {
         auto& h = c.get_histogram();
         auto mh = add_label(mf.add_metric(), id,ctx)->mutable_summary();
         mh->set_sample_count(h.sample_count);
@@ -177,7 +177,7 @@ static void fill_metric(pm::MetricFamily& mf, const metrics::impl::metric_value&
         mf.set_type(pm::MetricType::SUMMARY);
         break;
     }
-    case scollectd::data_type::HISTOGRAM:
+    case seastar::metrics::impl::data_type::HISTOGRAM:
     {
         auto& h = c.get_histogram();
         auto mh = add_label(mf.add_metric(), id,ctx)->mutable_histogram();
@@ -191,9 +191,9 @@ static void fill_metric(pm::MetricFamily& mf, const metrics::impl::metric_value&
         mf.set_type(pm::MetricType::HISTOGRAM);
         break;
     }
-    case scollectd::data_type::REAL_COUNTER:
+    case seastar::metrics::impl::data_type::REAL_COUNTER:
         [[fallthrough]];
-    case scollectd::data_type::COUNTER:
+    case seastar::metrics::impl::data_type::COUNTER:
         add_label(mf.add_metric(), id, ctx)->mutable_counter()->set_value(c.d());
         mf.set_type(pm::MetricType::COUNTER);
         break;
diff --git a/src/core/scollectd-impl.hh b/src/core/scollectd-impl.hh
deleted file mode 100644
index 94ea43e445..0000000000
--- a/src/core/scollectd-impl.hh
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * This file is open source software, licensed to you under the terms
- * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
- * distributed with this work for additional information regarding copyright
- * ownership.  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.
- */
-/*
- * Copyright (C) 2016 ScyllaDB
- */
-
-#pragma once
-
-#include <seastar/core/scollectd.hh>
-#include <seastar/core/metrics_api.hh>
-#include <seastar/net/api.hh>
-#include <chrono>
-
-namespace seastar {
-
-namespace scollectd {
-
-using namespace std::chrono_literals;
-using duration = std::chrono::milliseconds;
-
-static const ipv4_addr default_addr("239.192.74.66:25826");
-static const std::chrono::milliseconds default_period(1s);
-
-class impl {
-    net::datagram_channel _chan;
-    timer<> _timer;
-
-    sstring _host = "localhost";
-    ipv4_addr _addr = default_addr;
-    std::chrono::milliseconds _period = default_period;
-    uint64_t _num_packets = 0;
-    uint64_t _millis = 0;
-    uint64_t _bytes = 0;
-    double _avg = 0;
-
-public:
-    typedef seastar::metrics::impl::value_map value_list_map;
-    typedef value_list_map::value_type value_list_pair;
-
-    void add_polled(const type_instance_id & id,
-            const shared_ptr<value_list> & values, bool enable = true);
-    void remove_polled(const type_instance_id & id);
-    // explicitly send a type_instance value list (outside polling)
-    future<> send_metric(const type_instance_id & id,
-            const value_list & values);
-    future<> send_notification(const type_instance_id & id,
-            const sstring & msg);
-    // initiates actual value polling -> send to target "loop"
-    void start(const sstring & host, const ipv4_addr & addr, const std::chrono::milliseconds period);
-    void stop();
-
-    value_list_map& get_value_list_map();
-    const sstring& host() const {
-        return _host;
-    }
-
-private:
-    void arm();
-    void run();
-
-public:
-    shared_ptr<value_list> get_values(const type_instance_id & id) const;
-    std::vector<type_instance_id> get_instance_ids() const;
-    sstring get_collectd_description_str(const scollectd::type_instance_id&) const;
-private:
-    const value_list_map& values() const {
-        return seastar::metrics::impl::get_value_map();
-    }
-    seastar::metrics::metric_groups _metrics;
-};
-
-impl & get_impl();
-
-};
-
-}
diff --git a/src/core/scollectd.cc b/src/core/scollectd.cc
deleted file mode 100644
index 26110ce741..0000000000
--- a/src/core/scollectd.cc
+++ /dev/null
@@ -1,1001 +0,0 @@
-/*
- * This file is open source software, licensed to you under the terms
- * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
- * distributed with this work for additional information regarding copyright
- * ownership.  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.
- */
-/*
- * Copyright (C) 2014 Cloudius Systems, Ltd.
- */
-
-#ifdef SEASTAR_MODULE
-module;
-#endif
-
-#include <sys/socket.h>
-#include <cassert>
-#include <chrono>
-#include <cmath>
-#include <cstdint>
-#include <functional>
-#include <unordered_map>
-#include <utility>
-#include <string>
-#include <map>
-#include <iostream>
-#include <unordered_map>
-
-#ifdef SEASTAR_MODULE
-module seastar;
-#else
-#include <seastar/core/seastar.hh>
-#include <seastar/core/scollectd_api.hh>
-#include <seastar/core/metrics_api.hh>
-#include <seastar/core/byteorder.hh>
-#include <seastar/core/print.hh>
-#include <seastar/core/shared_ptr.hh>
-
-#include "core/scollectd-impl.hh"
-#endif
-
-namespace seastar {
-
-void scollectd::type_instance_id::truncate(sstring& field, const char* field_desc) {
-    if (field.size() > max_collectd_field_text_len) {
-        auto suffix_len = std::ceil(std::log10(++_next_truncated_idx)) + 1;
-        sstring new_field(seastar::format(
-            "{}~{:d}", sstring(field.data(), max_collectd_field_text_len - suffix_len), _next_truncated_idx));
-
-        logger.warn("Truncating \"{}\" to {} chars: \"{}\" -> \"{}\"", field_desc, max_collectd_field_text_len, field,
-            new_field);
-        field = std::move(new_field);
-    }
-}
-
-bool scollectd::type_instance_id::operator<(
-        const scollectd::type_instance_id& id2) const {
-    auto& id1 = *this;
-    return std::tie(id1.plugin(), id1.plugin_instance(), id1.type(),
-            id1.type_instance())
-            < std::tie(id2.plugin(), id2.plugin_instance(), id2.type(),
-                    id2.type_instance());
-}
-bool scollectd::type_instance_id::operator==(
-        const scollectd::type_instance_id & id2) const {
-    auto& id1 = *this;
-    return std::tie(id1.plugin(), id1.plugin_instance(), id1.type(),
-            id1.type_instance())
-            == std::tie(id2.plugin(), id2.plugin_instance(), id2.type(),
-                    id2.type_instance());
-}
-
-namespace scollectd {
-
-::seastar::logger logger("scollectd");
-thread_local unsigned type_instance_id::_next_truncated_idx = 0;
-
-registration::~registration() {
-    unregister();
-}
-
-registration::registration(const type_instance_id& id)
-: _id(id), _impl(seastar::metrics::impl::get_local_impl()) {
-}
-
-registration::registration(type_instance_id&& id)
-: _id(std::move(id)), _impl(seastar::metrics::impl::get_local_impl()) {
-}
-
-seastar::metrics::impl::metric_id to_metrics_id(const type_instance_id & id) {
-    seastar::metrics::impl::labels_type labels {{seastar::metrics::shard_label.name(), seastar::metrics::impl::shard()}};
-    auto internalized_labels = make_lw_shared<seastar::metrics::impl::labels_type>(std::move(labels));
-    return seastar::metrics::impl::metric_id(id.plugin(), id.type_instance(), std::move(internalized_labels));
-}
-
-
-const plugin_instance_id per_cpu_plugin_instance("#cpu");
-
-static const size_t payload_size = 1024;
-
-enum class part_type : uint16_t {
-    Host = 0x0000, // The name of the host to associate with subsequent data values
-    Time = 0x0001, // Time  Numeric The timestamp to associate with subsequent data values, unix time format (seconds since epoch)
-    TimeHr = 0x0008, // Time (high resolution)  Numeric The timestamp to associate with subsequent data values. Time is defined in 2–30 seconds since epoch. New in Version 5.0.
-    Plugin = 0x0002, // Plugin String The plugin name to associate with subsequent data values, e.g. "cpu"
-    PluginInst = 0x0003, // Plugin instance String  The plugin instance name to associate with subsequent data values, e.g. "1"
-    Type = 0x0004, // Type String The type name to associate with subsequent data values, e.g. "cpu"
-    TypeInst = 0x0005, // Type instance     String  The type instance name to associate with subsequent data values, e.g. "idle"
-    Values = 0x0006, // Values  other   Data values, see above
-    Interval = 0x0007, // Interval Numeric Interval used to set the "step" when creating new RRDs unless rrdtool plugin forces StepSize. Also used to detect values that have timed out.
-    IntervalHr = 0x0009, // Interval (high resolution)  Numeric     The interval in which subsequent data values are collected. The interval is given in 2–30 seconds. New in Version 5.0.
-    Message = 0x0100, // Message (notifications) String
-    Severity = 0x0101, // Severity  Numeric
-    Signature = 0x0200, // Signature (HMAC-SHA-256)     other (todo)
-    Encryption = 0x0210, // Encryption (AES-256/OFB
-};
-
-// "Time is defined in 2^–30 seconds since epoch. New in Version 5.0."
-typedef std::chrono::duration<uint64_t, std::ratio<1, 0x40000000>> collectd_hres_duration;
-
-// yet another writer type, this one to construct collectd network
-// protocol data.
-struct cpwriter {
-    typedef std::array<char, payload_size> buffer_type;
-    typedef buffer_type::iterator mark_type;
-    typedef buffer_type::const_iterator const_mark_type;
-
-    buffer_type _buf = {};
-    mark_type _pos;
-    bool _overflow = false;
-
-    std::unordered_map<uint16_t, sstring> _cache;
-
-    cpwriter()
-            : _pos(_buf.begin()) {
-    }
-    mark_type mark() const {
-        return _pos;
-    }
-    bool overflow() const {
-        return _overflow;
-    }
-    void reset(mark_type m) {
-        _pos = m;
-        _overflow = false;
-    }
-    size_t size() const {
-        return std::distance(_buf.begin(), const_mark_type(_pos));
-    }
-    bool empty() const {
-        return _pos == _buf.begin();
-    }
-    void clear() {
-        reset(_buf.begin());
-        _cache.clear();
-        _overflow = false;
-    }
-    const char * data() const {
-        return &_buf.at(0);
-    }
-    char * data() {
-        return &_buf.at(0);
-    }
-    cpwriter& check(size_t sz) {
-        size_t av = std::distance(_pos, _buf.end());
-        _overflow |= av < sz;
-        return *this;
-    }
-
-    sstring get_type_instance(const metrics::metric_name_type& name, const metrics::impl::labels_type& labels) {
-        if (labels.empty()) {
-            return name;
-        }
-        sstring res = name;
-        for (auto i : labels) {
-            if (i.first != seastar::metrics::shard_label.name()) {
-                res += "-" + i.second;
-            }
-        }
-        return res;
-    }
-    explicit operator bool() const {
-        return !_overflow;
-    }
-    bool operator!() const {
-        return !operator bool();
-    }
-    template<typename _Iter>
-    cpwriter & write(_Iter s, _Iter e) {
-        if (check(std::distance(s, e))) {
-            _pos = std::copy(s, e, _pos);
-        }
-        return *this;
-    }
-    template<typename T>
-    std::enable_if_t<std::is_integral_v<T>, cpwriter &> write(
-            const T & t) {
-        T tmp = net::hton(t);
-        auto * p = reinterpret_cast<const uint8_t *>(&tmp);
-        auto * e = p + sizeof(T);
-        write(p, e);
-        return *this;
-    }
-    template<typename T>
-    std::enable_if_t<std::is_integral_v<T>, cpwriter &> write_le(const T & t) {
-        T tmp = cpu_to_le(t);
-        auto * p = reinterpret_cast<const uint8_t *>(&tmp);
-        auto * e = p + sizeof(T);
-        write(p, e);
-        return *this;
-    }
-    void write_value(const seastar::metrics::impl::metric_value& v) {
-        switch (v.type()) {
-            case data_type::GAUGE: {
-                double tmpd = v.d();
-                uint64_t tmpi;
-                std::copy_n(reinterpret_cast<const char*>(&tmpd), 8, reinterpret_cast<char*>(&tmpi));
-                write_le(tmpi);
-                break;
-            }
-            case data_type::COUNTER:
-            case data_type::REAL_COUNTER:
-                write(v.ui()); // unsigned int 64, big endian
-                break;
-            default:
-                assert(0);
-        }
-    }
-    cpwriter & write(const sstring & s) {
-        write(s.begin(), s.end() + 1); // include \0
-        return *this;
-    }
-    cpwriter & put(part_type type, const sstring & s) {
-        write(uint16_t(type));
-        write(uint16_t(4 + s.size() + 1)); // include \0
-        write(s); // include \0
-        return *this;
-    }
-    cpwriter & put_cached(part_type type, const sstring & s) {
-        auto & cached = _cache[uint16_t(type)];
-        if (cached != s) {
-            put(type, s);
-            cached = s;
-        }
-        return *this;
-    }
-    template<typename T>
-    std::enable_if_t<std::is_integral_v<T>, cpwriter &> put(
-            part_type type, T t) {
-        write(uint16_t(type));
-        write(uint16_t(4 + sizeof(t)));
-        write(t);
-        return *this;
-    }
-    cpwriter & put(part_type type, const value_list & v) {
-        auto s = v.size();
-        auto sz = 6 + s + s * sizeof(uint64_t);
-        if (check(sz)) {
-            write(uint16_t(type));
-            write(uint16_t(sz));
-            write(uint16_t(s));
-            v.types(reinterpret_cast<data_type *>(&(*_pos)));
-            _pos += s;
-            v.values(reinterpret_cast<net::packed<uint64_t> *>(&(*_pos)));
-            _pos += s * sizeof(uint64_t);
-        }
-        return *this;
-    }
-
-    cpwriter & put(part_type type, const seastar::metrics::impl::metric_value & v) {
-        auto sz = 7 +  sizeof(uint64_t);
-        if (check(sz)) {
-            write(uint16_t(type));
-            write(uint16_t(sz));
-            write(uint16_t(1));
-            write(static_cast<uint8_t>(v.type()));
-            write_value(v);
-        }
-        return *this;
-    }
-    cpwriter & put(const sstring & host, const metrics::group_name_type& group_name, const metrics::metric_name_type& name,
-            const metrics::impl::labels_type& labels, const type_id& type) {
-        const auto ts = std::chrono::system_clock::now().time_since_epoch();
-        const auto lrts =
-                std::chrono::duration_cast<std::chrono::seconds>(ts).count();
-
-        put_cached(part_type::Host, host);
-        put(part_type::Time, uint64_t(lrts));
-        // Seems hi-res timestamp does not work very well with
-        // at the very least my default collectd in fedora (or I did it wrong?)
-        // Use lo-res ts for now, it is probably quite sufficient.
-        put_cached(part_type::Plugin, group_name);
-        // Optional
-        auto instance_id = labels.at(metrics::shard_label.name());
-        put_cached(part_type::PluginInst,
-                instance_id == per_cpu_plugin_instance ?
-                        to_sstring(this_shard_id()) : instance_id);
-        put_cached(part_type::Type, type);
-        // Optional
-        put_cached(part_type::TypeInst, get_type_instance(name, labels));
-        return *this;
-    }
-    cpwriter & put(const sstring & host,
-            const duration & period,
-            const type_instance_id & id, const value_list & v) {
-        const auto ps = std::chrono::duration_cast<collectd_hres_duration>(
-                        period).count();
-            auto mid = to_metrics_id(id);
-            put(host, mid.group_name(), mid.name(), mid.labels(), id.type());
-            put(part_type::Values, v);
-            if (ps != 0) {
-                put(part_type::IntervalHr, ps);
-            }
-            return *this;
-    }
-
-    cpwriter & put(const sstring & host,
-            const duration & period,
-            const type_id& type,
-            const metrics::group_name_type& group_name, const metrics::metric_name_type& name,
-            const metrics::impl::labels_type& labels, const seastar::metrics::impl::metric_value & v) {
-        const auto ps = std::chrono::duration_cast<collectd_hres_duration>(
-                period).count();
-        put(host, group_name, name, labels, type);
-        put(part_type::Values, v);
-        if (ps != 0) {
-            put(part_type::IntervalHr, ps);
-        }
-        return *this;
-    }
-};
-
-void impl::add_polled(const type_instance_id & id,
-        const shared_ptr<value_list> & values, bool enable) {
-    // do nothing
-    // add_polled is now implemented on the metrics layer
-
-}
-
-void impl::remove_polled(const type_instance_id & id) {
-    seastar::metrics::impl::unregister_metric(to_metrics_id(id));
-}
-
-// explicitly send a type_instance value list (outside polling)
-future<> impl::send_metric(const type_instance_id & id,
-        const value_list & values) {
-    if (values.empty()) {
-        return make_ready_future();
-    }
-    cpwriter out;
-    out.put(_host, duration(), id, values);
-    return _chan.send(_addr, net::packet(out.data(), out.size()));
-}
-
-future<> impl::send_notification(const type_instance_id & id,
-        const sstring & msg) {
-    cpwriter out;
-    auto mid = to_metrics_id(id);
-    out.put(_host, mid.group_name(), mid.name(), mid.labels(), id.type());
-    out.put(part_type::Message, msg);
-    return _chan.send(_addr, net::packet(out.data(), out.size()));
-}
-
-// initiates actual value polling -> send to target "loop"
-void impl::start(const sstring & host, const ipv4_addr & addr, const duration period) {
-    _period = period;
-    _addr = addr;
-    _host = host;
-    _chan = make_unbound_datagram_channel(AF_INET);
-    _timer.set_callback(std::bind(&impl::run, this));
-
-    // dogfood ourselves
-    namespace sm = seastar::metrics;
-
-    _metrics.add_group("scollectd", {
-        // total_bytes      value:DERIVE:0:U
-        sm::make_counter("total_bytes_sent", sm::description("total bytes sent"), _bytes),
-        // total_requests      value:DERIVE:0:U
-        sm::make_counter("total_requests", sm::description("total requests"), _num_packets),
-        // latency          value:GAUGE:0:U
-        sm::make_gauge("latency", sm::description("avrage latency"), _avg),
-        // total_time_in_ms    value:DERIVE:0:U
-        sm::make_counter("total_time_in_ms", sm::description("total time in milliseconds"), _millis),
-        // total_values     value:DERIVE:0:U
-        sm::make_gauge("total_values", sm::description("current number of values reported"), [this] {return values().size();}),
-        // records          value:GAUGE:0:U
-        sm::make_gauge("records", sm::description("number of records reported"), [this] {return values().size();}),
-    });
-
-    // FIXME: future is discarded
-    (void)send_notification(
-            type_instance_id("scollectd", per_cpu_plugin_instance,
-                    "network"), "daemon started");
-    arm();
-}
-
-void impl::stop() {
-    _timer.cancel();
-    _metrics.clear();
-}
-
-
-void impl::arm() {
-    if (_period != duration()) {
-        _timer.arm(_period);
-    }
-}
-
-void impl::run() {
-    typedef size_t metric_family_id;
-    typedef seastar::metrics::impl::value_vector::iterator value_iterator;
-    typedef seastar::metrics::impl::metric_metadata_fifo::iterator metadata_iterator;
-    typedef std::tuple<metric_family_id, metadata_iterator, value_iterator, type_id, cpwriter> context;
-
-    auto ctxt = make_lw_shared<context>();
-    foreign_ptr<shared_ptr<seastar::metrics::impl::values_copy>> vals =  seastar::metrics::impl::get_values();
-
-    // note we're doing this unsynced since we assume
-    // all registrations to this instance will be done on the
-    // same cpu, and without interuptions (no wait-states)
-
-    auto& values = vals->values;
-    auto metadata = vals->metadata;
-    std::get<metric_family_id>(*ctxt) = 0;
-    if (values.size() > 0) {
-        std::get<value_iterator>(*ctxt) = values[0].begin();
-        std::get<metadata_iterator>(*ctxt) = metadata->at(0).metrics.begin();
-        std::get<type_id>(*ctxt) = metadata->at(0).mf.inherit_type;
-    }
-
-    auto stop_when = [ctxt, metadata]() {
-        auto done = std::get<metric_family_id>(*ctxt) == metadata->size();
-        return done;
-    };
-    // append as many values as we can fit into a packet (1024 bytes)
-    auto send_packet = [this, ctxt, &values, metadata]() mutable {
-        auto start = steady_clock_type::now();
-        auto& mf = std::get<metric_family_id>(*ctxt);
-        auto & md_iterator = std::get<metadata_iterator>(*ctxt);
-        auto & i = std::get<value_iterator>(*ctxt);
-        auto & out = std::get<cpwriter>(*ctxt);
-
-        out.clear();
-
-        bool out_of_space = false;
-        while (!out_of_space && mf < values.size()) {
-            while (i != values[mf].end()) {
-                if (i->type() == seastar::metrics::impl::data_type::HISTOGRAM) {
-                    ++i;
-                    ++md_iterator;
-                    continue;
-                }
-                auto m = out.mark();
-                out.put(_host, _period, std::get<type_id>(*ctxt),
-                    md_iterator->group_name(), md_iterator->name(), md_iterator->labels(), *i);
-                if (!out) {
-                    out.reset(m);
-                    out_of_space = true;
-                    break;
-                }
-                ++i;
-                ++md_iterator;
-            }
-            if (out_of_space) {
-                break;
-            }
-            ++mf;
-            if (mf < values.size()) {
-                i = values[mf].begin();
-                md_iterator = metadata->at(mf).metrics.begin();
-                std::get<type_id>(*ctxt) = metadata->at(mf).mf.inherit_type;
-            }
-        }
-        if (out.empty()) {
-            return make_ready_future();
-        }
-        return _chan.send(_addr, net::packet(out.data(), out.size())).then([start, ctxt, this]() {
-                    auto & out = std::get<cpwriter>(*ctxt);
-                    auto now = steady_clock_type::now();
-                    // dogfood stats
-                    ++_num_packets;
-                    _millis += std::chrono::duration_cast<std::chrono::milliseconds>(now - start).count();
-                    _bytes += out.size();
-                    _avg = double(_millis) / _num_packets;
-                }).then_wrapped([] (auto&& f) {
-                    try {
-                        f.get();
-                    } catch (std::exception & ex) {
-                        std::cout << "send failed: " << ex.what() << std::endl;
-                    } catch (...) {
-                        std::cout << "send failed: - unknown exception" << std::endl;
-                    }
-                });
-    };
-    // No need to wait for future.
-    // The caller has to call impl::stop() to synchronize.
-    (void)do_until(stop_when, send_packet).finally([this, vals = std::move(vals)]() mutable {
-        arm();
-    });
-}
-
-std::vector<type_instance_id> impl::get_instance_ids() const {
-    std::vector<type_instance_id> res;
-    for (auto&& v: values()) {
-        // Need to check for empty value_list, since unreg is two-stage.
-        // Not an issue for most uses, but unit testing etc that would like
-        // fully deterministic operation here would like us to only return
-        // actually active ids
-        for (auto i : v.second) {
-            if (i.second) {
-                res.emplace_back(i.second->get_id(), v.second.info().inherit_type);
-            }
-        }
-    }
-    return res;
-}
-
-void add_polled(const type_instance_id & id,
-        const shared_ptr<value_list> & values, bool enabled) {
-    get_impl().add_polled(id, values, enabled);
-}
-
-void remove_polled_metric(const type_instance_id & id) {
-    get_impl().remove_polled(id);
-}
-
-future<> send_notification(const type_instance_id & id,
-        const sstring & msg) {
-    return get_impl().send_notification(id, msg);
-}
-
-future<> send_metric(const type_instance_id & id,
-        const value_list & values) {
-    return get_impl().send_metric(id, values);
-}
-
-void configure(const options& opts) {
-    bool enable = opts.collectd.get_value();
-    if (!enable) {
-        return;
-    }
-    auto addr = ipv4_addr(opts.collectd_address.get_value());
-    auto period = std::chrono::milliseconds(opts.collectd_poll_period.get_value());
-
-    auto host = (opts.collectd_hostname.get_value() == "")
-            ? seastar::metrics::impl::get_local_impl()->get_config().hostname
-            : sstring(opts.collectd_hostname.get_value());
-
-    // Now create send loops on each cpu
-    for (unsigned c = 0; c < smp::count; c++) {
-        // FIXME: future is discarded
-        (void)smp::submit_to(c, [=] () {
-            get_impl().start(host, addr, period);
-        });
-    }
-}
-
-options::options(program_options::option_group* parent_group)
-    : program_options::option_group(parent_group, "COLLECTD options")
-    , collectd(*this, "collectd", false,
-            "enable collectd daemon")
-    , collectd_address(*this, "collectd-address",
-            "239.192.74.66:25826",
-            "address to send/broadcast metrics to")
-    , collectd_poll_period(*this, "collectd-poll-period",
-            1000,
-            "poll period - frequency of sending counter metrics (default: 1000ms, 0 disables)")
-    , collectd_hostname(*this, "collectd-hostname",
-            "",
-            "Deprecated option, use metrics-hostname instead")
-{
-}
-
-static seastar::metrics::impl::register_ref get_register(const scollectd::type_instance_id& i) {
-    seastar::metrics::impl::metric_id id = to_metrics_id(i);
-    return seastar::metrics::impl::get_value_map().at(id.full_name()).at(id.internalized_labels());
-}
-
-std::vector<collectd_value> get_collectd_value(
-        const scollectd::type_instance_id& id) {
-    std::vector<collectd_value> vals;
-    const seastar::metrics::impl::registered_metric& val = *get_register(id);
-    vals.push_back(val());
-    return vals;
-}
-
-std::vector<scollectd::type_instance_id> get_collectd_ids() {
-    return get_impl().get_instance_ids();
-}
-
-bool is_enabled(const scollectd::type_instance_id& id) {
-    return get_register(id)->is_enabled();
-}
-
-void enable(const scollectd::type_instance_id& id, bool enable) {
-    get_register(id)->set_enabled(enable);
-}
-
-type_instance_id plugin_instance_metrics::add_impl(const typed_value& v) {
-    type_instance_id id(_plugin_id, _plugin_instance, v.type(), v.type_instance());
-    get_impl().add_polled(id, v.values());
-    return id;
-}
-
-void plugin_instance_metrics::add(const typed_value& v) {
-    _registrations.emplace_back(add_impl(v));
-}
-
-std::vector<type_instance_id> plugin_instance_metrics::bound_ids() const {
-    std::vector<type_instance_id> res;
-    res.reserve(_registrations.size());
-    std::transform(_registrations.begin(), _registrations.end(), std::back_inserter(res), [](const registration& r) {
-       return r._id;
-    });
-    return res;
-}
-
-type_id type_id_for(known_type t) {
-    switch (t) {
-    case known_type::absolute:
-        return "absolute";
-    case known_type::backends:
-        return "backends";
-    case known_type::bitrate:
-        return "bitrate";
-    case known_type::blocked_clients:
-        return "blocked_clients";
-    case known_type::bytes:
-        return "bytes";
-    case known_type::cache_eviction:
-        return "cache_eviction";
-    case known_type::cache_operation:
-        return "cache_operation";
-    case known_type::cache_ratio:
-        return "cache_ratio";
-    case known_type::cache_result:
-        return "cache_result";
-    case known_type::cache_size:
-        return "cache_size";
-    case known_type::capacity:
-        return "capacity";
-    case known_type::changes_since_last_save:
-        return "changes_since_last_save";
-    case known_type::charge:
-        return "charge";
-    case known_type::clock_last_meas:
-        return "clock_last_meas";
-    case known_type::clock_last_update:
-        return "clock_last_update";
-    case known_type::clock_mode:
-        return "clock_mode";
-    case known_type::clock_reachability:
-        return "clock_reachability";
-    case known_type::clock_skew_ppm:
-        return "clock_skew_ppm";
-    case known_type::clock_state:
-        return "clock_state";
-    case known_type::clock_stratum:
-        return "clock_stratum";
-    case known_type::compression:
-        return "compression";
-    case known_type::compression_ratio:
-        return "compression_ratio";
-    case known_type::connections:
-        return "connections";
-    case known_type::conntrack:
-        return "conntrack";
-    case known_type::contextswitch:
-        return "contextswitch";
-    case known_type::count:
-        return "count";
-    case known_type::counter:
-        return "counter";
-    case known_type::cpu:
-        return "cpu";
-    case known_type::cpufreq:
-        return "cpufreq";
-    case known_type::current:
-        return "current";
-    case known_type::current_connections:
-        return "current_connections";
-    case known_type::current_sessions:
-        return "current_sessions";
-    case known_type::delay:
-        return "delay";
-    case known_type::derive:
-        return "derive";
-    case known_type::df:
-        return "df";
-    case known_type::df_complex:
-        return "df_complex";
-    case known_type::df_inodes:
-        return "df_inodes";
-    case known_type::disk_io_time:
-        return "disk_io_time";
-    case known_type::disk_latency:
-        return "disk_latency";
-    case known_type::disk_merged:
-        return "disk_merged";
-    case known_type::disk_octets:
-        return "disk_octets";
-    case known_type::disk_ops:
-        return "disk_ops";
-    case known_type::disk_ops_complex:
-        return "disk_ops_complex";
-    case known_type::disk_time:
-        return "disk_time";
-    case known_type::dns_answer:
-        return "dns_answer";
-    case known_type::dns_notify:
-        return "dns_notify";
-    case known_type::dns_octets:
-        return "dns_octets";
-    case known_type::dns_opcode:
-        return "dns_opcode";
-    case known_type::dns_qtype:
-        return "dns_qtype";
-    case known_type::dns_qtype_cached:
-        return "dns_qtype_cached";
-    case known_type::dns_query:
-        return "dns_query";
-    case known_type::dns_question:
-        return "dns_question";
-    case known_type::dns_rcode:
-        return "dns_rcode";
-    case known_type::dns_reject:
-        return "dns_reject";
-    case known_type::dns_request:
-        return "dns_request";
-    case known_type::dns_resolver:
-        return "dns_resolver";
-    case known_type::dns_response:
-        return "dns_response";
-    case known_type::dns_transfer:
-        return "dns_transfer";
-    case known_type::dns_update:
-        return "dns_update";
-    case known_type::dns_zops:
-        return "dns_zops";
-    case known_type::drbd_resource:
-        return "drbd_resource";
-    case known_type::duration:
-        return "duration";
-    case known_type::email_check:
-        return "email_check";
-    case known_type::email_count:
-        return "email_count";
-    case known_type::email_size:
-        return "email_size";
-    case known_type::entropy:
-        return "entropy";
-    case known_type::evicted_keys:
-        return "evicted_keys";
-    case known_type::expired_keys:
-        return "expired_keys";
-    case known_type::fanspeed:
-        return "fanspeed";
-    case known_type::file_handles:
-        return "file_handles";
-    case known_type::file_size:
-        return "file_size";
-    case known_type::files:
-        return "files";
-    case known_type::flow:
-        return "flow";
-    case known_type::fork_rate:
-        return "fork_rate";
-    case known_type::frequency:
-        return "frequency";
-    case known_type::frequency_error:
-        return "frequency_error";
-    case known_type::frequency_offset:
-        return "frequency_offset";
-    case known_type::fscache_stat:
-        return "fscache_stat";
-    case known_type::gauge:
-        return "gauge";
-    case known_type::hash_collisions:
-        return "hash_collisions";
-    case known_type::http_request_methods:
-        return "http_request_methods";
-    case known_type::http_requests:
-        return "http_requests";
-    case known_type::http_response_codes:
-        return "http_response_codes";
-    case known_type::humidity:
-        return "humidity";
-    case known_type::if_collisions:
-        return "if_collisions";
-    case known_type::if_dropped:
-        return "if_dropped";
-    case known_type::if_errors:
-        return "if_errors";
-    case known_type::if_multicast:
-        return "if_multicast";
-    case known_type::if_octets:
-        return "if_octets";
-    case known_type::if_packets:
-        return "if_packets";
-    case known_type::if_rx_errors:
-        return "if_rx_errors";
-    case known_type::if_rx_octets:
-        return "if_rx_octets";
-    case known_type::if_tx_errors:
-        return "if_tx_errors";
-    case known_type::if_tx_octets:
-        return "if_tx_octets";
-    case known_type::invocations:
-        return "invocations";
-    case known_type::io_octets:
-        return "io_octets";
-    case known_type::io_packets:
-        return "io_packets";
-    case known_type::ipt_bytes:
-        return "ipt_bytes";
-    case known_type::ipt_packets:
-        return "ipt_packets";
-    case known_type::irq:
-        return "irq";
-    case known_type::latency:
-        return "latency";
-    case known_type::links:
-        return "links";
-    case known_type::load:
-        return "load";
-    case known_type::md_disks:
-        return "md_disks";
-    case known_type::memory:
-        return "memory";
-    case known_type::memory_lua:
-        return "memory_lua";
-    case known_type::memory_throttle_count:
-        return "memory_throttle_count";
-    case known_type::multimeter:
-        return "multimeter";
-    case known_type::mutex_operations:
-        return "mutex_operations";
-    case known_type::objects:
-        return "objects";
-    case known_type::operations:
-        return "operations";
-    case known_type::packets:
-        return "packets";
-    case known_type::pending_operations:
-        return "pending_operations";
-    case known_type::percent:
-        return "percent";
-    case known_type::percent_bytes:
-        return "percent_bytes";
-    case known_type::percent_inodes:
-        return "percent_inodes";
-    case known_type::ping:
-        return "ping";
-    case known_type::ping_droprate:
-        return "ping_droprate";
-    case known_type::ping_stddev:
-        return "ping_stddev";
-    case known_type::players:
-        return "players";
-    case known_type::power:
-        return "power";
-    case known_type::pressure:
-        return "pressure";
-    case known_type::protocol_counter:
-        return "protocol_counter";
-    case known_type::pubsub:
-        return "pubsub";
-    case known_type::queue_length:
-        return "queue_length";
-    case known_type::records:
-        return "records";
-    case known_type::requests:
-        return "requests";
-    case known_type::response_code:
-        return "response_code";
-    case known_type::response_time:
-        return "response_time";
-    case known_type::root_delay:
-        return "root_delay";
-    case known_type::root_dispersion:
-        return "root_dispersion";
-    case known_type::route_etx:
-        return "route_etx";
-    case known_type::route_metric:
-        return "route_metric";
-    case known_type::routes:
-        return "routes";
-    case known_type::segments:
-        return "segments";
-    case known_type::serial_octets:
-        return "serial_octets";
-    case known_type::signal_noise:
-        return "signal_noise";
-    case known_type::signal_power:
-        return "signal_power";
-    case known_type::signal_quality:
-        return "signal_quality";
-    case known_type::snr:
-        return "snr";
-    case known_type::spl:
-        return "spl";
-    case known_type::swap:
-        return "swap";
-    case known_type::swap_io:
-        return "swap_io";
-    case known_type::tcp_connections:
-        return "tcp_connections";
-    case known_type::temperature:
-        return "temperature";
-    case known_type::threads:
-        return "threads";
-    case known_type::time_dispersion:
-        return "time_dispersion";
-    case known_type::time_offset:
-        return "time_offset";
-    case known_type::time_offset_ntp:
-        return "time_offset_ntp";
-    case known_type::time_offset_rms:
-        return "time_offset_rms";
-    case known_type::time_ref:
-        return "time_ref";
-    case known_type::timeleft:
-        return "timeleft";
-    case known_type::total_bytes:
-        return "total_bytes";
-    case known_type::total_connections:
-        return "total_connections";
-    case known_type::total_objects:
-        return "total_objects";
-    case known_type::total_operations:
-        return "total_operations";
-    case known_type::total_requests:
-        return "total_requests";
-    case known_type::total_sessions:
-        return "total_sessions";
-    case known_type::total_threads:
-        return "total_threads";
-    case known_type::total_time_in_ms:
-        return "total_time_in_ms";
-    case known_type::total_values:
-        return "total_values";
-    case known_type::uptime:
-        return "uptime";
-    case known_type::users:
-        return "users";
-    case known_type::vcl:
-        return "vcl";
-    case known_type::vcpu:
-        return "vcpu";
-    case known_type::virt_cpu_total:
-        return "virt_cpu_total";
-    case known_type::virt_vcpu:
-        return "virt_vcpu";
-    case known_type::vmpage_action:
-        return "vmpage_action";
-    case known_type::vmpage_faults:
-        return "vmpage_faults";
-    case known_type::vmpage_io:
-        return "vmpage_io";
-    case known_type::vmpage_number:
-        return "vmpage_number";
-    case known_type::volatile_changes:
-        return "volatile_changes";
-    case known_type::voltage:
-        return "voltage";
-    case known_type::voltage_threshold:
-        return "voltage_threshold";
-    case known_type::vs_memory:
-        return "vs_memory";
-    case known_type::vs_processes:
-        return "vs_processes";
-    case known_type::vs_threads:
-        return "vs_threads";
-    default:
-        throw std::invalid_argument("Unknown type");
-    }
-}
-
-metrics::impl::value_map get_value_map() {
-    return metrics::impl::get_value_map();
-}
-
-}
-
-thread_local scollectd::impl scollectd_impl;
-
-scollectd::impl & scollectd::get_impl() {
-    return scollectd_impl;
-}
-
-}
diff --git a/src/seastar.cc b/src/seastar.cc
index 29295518f3..0aa1300fd9 100644
--- a/src/seastar.cc
+++ b/src/seastar.cc
@@ -226,8 +226,6 @@ export module seastar;
 #include <seastar/core/scattered_message.hh>
 #include <seastar/core/scheduling.hh>
 #include <seastar/core/scheduling_specific.hh>
-#include <seastar/core/scollectd.hh>
-#include <seastar/core/scollectd_api.hh>
 #include <seastar/core/seastar.hh>
 #include <seastar/core/semaphore.hh>
 #include <seastar/core/sharded.hh>
@@ -318,7 +316,6 @@ module : private;
 #include "core/reactor_backend.hh"
 #include "core/syscall_result.hh"
 #include "core/thread_pool.hh"
-#include "core/scollectd-impl.hh"
 #include "core/vla.hh"
 
 #include <seastar/util/internal/iovec_utils.hh>