[ntcore] Add DataLog support

ntcore/src/main/java/edu/wpi/first/networktables/NetworkTableInstance.java

@@ -4,6 +4,7 @@
 
 package edu.wpi.first.networktables;
 
+import edu.wpi.first.util.datalog.DataLog;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -997,6 +998,50 @@ public final class NetworkTableInstance implements AutoCloseable {
     return NetworkTablesJNI.loadEntries(m_handle, filename, prefix);
   }
 
+  /**
+   * Starts logging entry changes to a DataLog.
+   *
+   * @param log data log object; lifetime must extend until StopEntryDataLog is called or the
+   *     instance is destroyed
+   * @param prefix only store entries with names that start with this prefix; the prefix is not
+   *     included in the data log entry name
+   * @param logPrefix prefix to add to data log entry names
+   * @return Data logger handle
+   */
+  public int startEntryDataLog(DataLog log, String prefix, String logPrefix) {
+    return NetworkTablesJNI.startEntryDataLog(m_handle, log, prefix, logPrefix);
+  }
+
+  /**
+   * Stops logging entry changes to a DataLog.
+   *
+   * @param logger data logger handle
+   */
+  public static void stopEntryDataLog(int logger) {
+    NetworkTablesJNI.stopEntryDataLog(logger);
+  }
+
+  /**
+   * Starts logging connection changes to a DataLog.
+   *
+   * @param log data log object; lifetime must extend until StopConnectionDataLog is called or the
+   *     instance is destroyed
+   * @param name data log entry name
+   * @return Data logger handle
+   */
+  public int startConnectionDataLog(DataLog log, String name) {
+    return NetworkTablesJNI.startConnectionDataLog(m_handle, log, name);
+  }
+
+  /**
+   * Stops logging connection changes to a DataLog.
+   *
+   * @param logger data logger handle
+   */
+  public static void stopConnectionDataLog(int logger) {
+    NetworkTablesJNI.stopConnectionDataLog(logger);
+  }
+
   private final ReentrantLock m_loggerLock = new ReentrantLock();
   private final Map<Integer, Consumer<LogMessage>> m_loggers = new HashMap<>();
   private int m_loggerPoller;

ntcore/src/main/java/edu/wpi/first/networktables/NetworkTablesJNI.java

@@ -5,6 +5,7 @@
 package edu.wpi.first.networktables;
 
 import edu.wpi.first.util.RuntimeLoader;
+import edu.wpi.first.util.datalog.DataLog;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -255,6 +256,22 @@ public final class NetworkTablesJNI {
 
   public static native long now();
 
+  private static native int startEntryDataLog(int inst, long log, String prefix, String logPrefix);
+
+  public static int startEntryDataLog(int inst, DataLog log, String prefix, String logPrefix) {
+    return startEntryDataLog(inst, log.getImpl(), prefix, logPrefix);
+  }
+
+  public static native void stopEntryDataLog(int logger);
+
+  private static native int startConnectionDataLog(int inst, long log, String name);
+
+  public static int startConnectionDataLog(int inst, DataLog log, String name) {
+    return startConnectionDataLog(inst, log.getImpl(), name);
+  }
+
+  public static native void stopConnectionDataLog(int logger);
+
   public static native int createLoggerPoller(int inst);
 
   public static native void destroyLoggerPoller(int poller);

ntcore/src/main/native/cpp/ConnectionNotifier.cpp

@@ -21,6 +21,10 @@ unsigned int ConnectionNotifier::AddPolled(unsigned int poller_uid) {
   return DoAdd(poller_uid);
 }
 
+void ConnectionNotifier::Remove(unsigned int uid) {
+  CallbackManager::Remove(uid);
+}
+
 void ConnectionNotifier::NotifyConnection(bool connected,
                                           const ConnectionInfo& conn_info,
                                           unsigned int only_listener) {

ntcore/src/main/native/cpp/ConnectionNotifier.h

@@ -63,6 +63,8 @@ class ConnectionNotifier
                        callback) override;
   unsigned int AddPolled(unsigned int poller_uid) override;
 
+  void Remove(unsigned int uid) override;
+
   void NotifyConnection(bool connected, const ConnectionInfo& conn_info,
                         unsigned int only_listener = UINT_MAX) override;
 

ntcore/src/main/native/cpp/Dispatcher.cpp

@@ -11,6 +11,8 @@
 #include <wpi/StringExtras.h>
 #include <wpi/TCPAcceptor.h>
 #include <wpi/TCPConnector.h>
+#include <wpi/json_serializer.h>
+#include <wpi/raw_ostream.h>
 #include <wpi/timestamp.h>
 
 #include "IConnectionNotifier.h"
@@ -20,6 +22,24 @@
 
 using namespace nt;
 
+static std::string ConnInfoToJson(bool connected, const ConnectionInfo& info) {
+  std::string str;
+  wpi::raw_string_ostream os{str};
+  wpi::json::serializer s{os, ' ', 0};
+  os << "{\"connected\":" << (connected ? "true" : "false");
+  os << ",\"remote_id\":\"";
+  s.dump_escaped(info.remote_id, false);
+  os << "\",\"remote_ip\":\"";
+  s.dump_escaped(info.remote_ip, false);
+  os << "\",\"remote_port\":";
+  s.dump_integer(static_cast<uint64_t>(info.remote_port));
+  os << ",\"protocol_version\":";
+  s.dump_integer(static_cast<uint64_t>(info.protocol_version));
+  os << "}";
+  os.flush();
+  return str;
+}
+
 void Dispatcher::StartServer(std::string_view persist_filename,
                              const char* listen_address, unsigned int port) {
   std::string listen_address_copy(wpi::trim(listen_address));
@@ -101,6 +121,13 @@ DispatcherBase::DispatcherBase(IStorage& storage, IConnectionNotifier& notifier,
 
 DispatcherBase::~DispatcherBase() {
   Stop();
+
+  {
+    std::scoped_lock lock(m_user_mutex);
+    for (auto&& datalog : m_dataloggers) {
+      m_notifier.Remove(datalog.notifier);
+    }
+  }
 }
 
 unsigned int DispatcherBase::GetNetworkMode() const {
@@ -302,6 +329,33 @@ unsigned int DispatcherBase::AddPolledListener(unsigned int poller_uid,
   return uid;
 }
 
+unsigned int DispatcherBase::StartDataLog(wpi::log::DataLog& log,
+                                          std::string_view name) {
+  std::scoped_lock lock(m_user_mutex);
+  auto now = nt::Now();
+  unsigned int uid = m_dataloggers.emplace_back(log, name, now);
+  m_dataloggers[uid].notifier =
+      m_notifier.Add([this, uid](const ConnectionNotification& n) {
+        std::scoped_lock lock(m_user_mutex);
+        if (uid < m_dataloggers.size() && m_dataloggers[uid].entry) {
+          m_dataloggers[uid].entry.Append(ConnInfoToJson(n.connected, n.conn),
+                                          nt::Now());
+        }
+      });
+  for (auto& conn : m_connections) {
+    if (conn->state() != NetworkConnection::kActive) {
+      continue;
+    }
+    m_dataloggers[uid].entry.Append(ConnInfoToJson(true, conn->info()), now);
+  }
+  return uid;
+}
+
+void DispatcherBase::StopDataLog(unsigned int logger) {
+  std::scoped_lock lock(m_user_mutex);
+  m_notifier.Remove(m_dataloggers.erase(logger).notifier);
+}
+
 void DispatcherBase::SetConnector(Connector connector) {
   std::scoped_lock lock(m_user_mutex);
   m_client_connector = std::move(connector);

ntcore/src/main/native/cpp/Dispatcher.h

@@ -14,6 +14,8 @@
 #include <utility>
 #include <vector>
 
+#include <wpi/DataLog.h>
+#include <wpi/UidVector.h>
 #include <wpi/condition_variable.h>
 #include <wpi/mutex.h>
 #include <wpi/span.h>
@@ -61,6 +63,9 @@ class DispatcherBase : public IDispatcher {
   unsigned int AddPolledListener(unsigned int poller_uid,
                                  bool immediate_notify) const;
 
+  unsigned int StartDataLog(wpi::log::DataLog& log, std::string_view name);
+  void StopDataLog(unsigned int logger);
+
   void SetConnector(Connector connector);
   void SetConnectorOverride(Connector connector);
   void ClearConnectorOverride();
@@ -120,6 +125,20 @@ class DispatcherBase : public IDispatcher {
   unsigned int m_reconnect_proto_rev = 0x0300;
   bool m_do_reconnect = true;
 
+  struct DataLogger {
+    DataLogger() = default;
+    DataLogger(wpi::log::DataLog& log, std::string_view name, int64_t time)
+        : entry{log, name,
+                "{\"schema\":\"NTConnectionInfo\",\"source\":\"NT\"}", "json",
+                time} {}
+
+    explicit operator bool() const { return static_cast<bool>(entry); }
+
+    wpi::log::StringLogEntry entry;
+    unsigned int notifier = 0;
+  };
+  wpi::UidVector<DataLogger, 4> m_dataloggers;
+
  protected:
   wpi::Logger& m_logger;
 };

ntcore/src/main/native/cpp/Handle.h

@@ -28,7 +28,9 @@ class Handle {
     kLogger,
     kLoggerPoller,
     kRpcCall,
-    kRpcCallPoller
+    kRpcCallPoller,
+    kDataLogger,
+    kConnectionDataLogger
   };
   enum { kIndexMax = 0xfffff };
 

ntcore/src/main/native/cpp/IConnectionNotifier.h

@@ -20,6 +20,7 @@ class IConnectionNotifier {
   virtual unsigned int Add(
       std::function<void(const ConnectionNotification& event)> callback) = 0;
   virtual unsigned int AddPolled(unsigned int poller_uid) = 0;
+  virtual void Remove(unsigned int uid) = 0;
   virtual void NotifyConnection(bool connected, const ConnectionInfo& conn_info,
                                 unsigned int only_listener = UINT_MAX) = 0;
 };

ntcore/src/main/native/cpp/Storage.cpp

@@ -4,6 +4,7 @@
 
 #include "Storage.h"
 
+#include <wpi/DataLog.h>
 #include <wpi/StringExtras.h>
 #include <wpi/timestamp.h>
 
@@ -13,6 +14,7 @@
 #include "INetworkConnection.h"
 #include "IRpcServer.h"
 #include "Log.h"
+#include "ntcore_c.h"
 
 using namespace nt;
 
@@ -144,8 +146,7 @@ void Storage::ProcessIncomingEntryAssign(std::shared_ptr<Message> msg,
         entry->seq_num = seq_num;
 
         // notify
-        m_notifier.NotifyEntry(entry->local_id, name, entry->value,
-                               NT_NOTIFY_NEW);
+        Notify(entry, NT_NOTIFY_NEW, false);
         return;
       }
       may_need_update = true;  // we may need to send an update message
@@ -208,7 +209,7 @@ void Storage::ProcessIncomingEntryAssign(std::shared_ptr<Message> msg,
   entry->seq_num = seq_num;
 
   // notify
-  m_notifier.NotifyEntry(entry->local_id, name, entry->value, notify_flags);
+  Notify(entry, notify_flags, false);
 
   // broadcast to all other connections (note for client there won't
   // be any other connections, so don't bother)
@@ -250,8 +251,7 @@ void Storage::ProcessIncomingEntryUpdate(std::shared_ptr<Message> msg,
   }
 
   // notify
-  m_notifier.NotifyEntry(entry->local_id, entry->name, entry->value,
-                         NT_NOTIFY_UPDATE);
+  Notify(entry, NT_NOTIFY_UPDATE, false);
 
   // broadcast to all other connections (note for client there won't
   // be any other connections, so don't bother)
@@ -453,8 +453,7 @@ void Storage::ApplyInitialAssignments(
       entry->value = msg->value();
       entry->flags = msg->flags();
       // notify
-      m_notifier.NotifyEntry(entry->local_id, name, entry->value,
-                             NT_NOTIFY_NEW);
+      Notify(entry, NT_NOTIFY_NEW, false);
     } else {
       // if we have written the value locally and the value is not persistent,
       // then we don't update the local value and instead send it back to the
@@ -474,8 +473,7 @@ void Storage::ApplyInitialAssignments(
           entry->flags = msg->flags();
         }
         // notify
-        m_notifier.NotifyEntry(entry->local_id, name, entry->value,
-                               notify_flags);
+        Notify(entry, notify_flags, false);
       }
     }
 
@@ -628,11 +626,9 @@ void Storage::SetEntryValueImpl(Entry* entry, std::shared_ptr<Value> value,
 
   // notify
   if (!old_value) {
-    m_notifier.NotifyEntry(entry->local_id, entry->name, value,
-                           NT_NOTIFY_NEW | (local ? NT_NOTIFY_LOCAL : 0));
+    Notify(entry, NT_NOTIFY_NEW, local);
   } else if (*old_value != *value) {
-    m_notifier.NotifyEntry(entry->local_id, entry->name, value,
-                           NT_NOTIFY_UPDATE | (local ? NT_NOTIFY_LOCAL : 0));
+    Notify(entry, NT_NOTIFY_UPDATE, local);
   }
 
   // remember local changes
@@ -732,8 +728,7 @@ void Storage::SetEntryFlagsImpl(Entry* entry, unsigned int flags,
   entry->flags = flags;
 
   // notify
-  m_notifier.NotifyEntry(entry->local_id, entry->name, entry->value,
-                         NT_NOTIFY_FLAGS | (local ? NT_NOTIFY_LOCAL : 0));
+  Notify(entry, NT_NOTIFY_FLAGS, local);
 
   // generate message
   if (!local || !m_dispatcher) {
@@ -817,8 +812,7 @@ void Storage::DeleteEntryImpl(Entry* entry, std::unique_lock<wpi::mutex>& lock,
   }
 
   // notify
-  m_notifier.NotifyEntry(entry->local_id, entry->name, old_value,
-                         NT_NOTIFY_DELETE | (local ? NT_NOTIFY_LOCAL : 0));
+  Notify(entry, NT_NOTIFY_DELETE, local, old_value);
 
   // if it had a value, generate message
   // don't send an update if we don't have an assigned id yet
@@ -832,14 +826,160 @@ void Storage::DeleteEntryImpl(Entry* entry, std::unique_lock<wpi::mutex>& lock,
   }
 }
 
+static std::string_view GetStorageTypeStr(NT_Type type) {
+  switch (type) {
+    case NT_BOOLEAN:
+      return wpi::log::BooleanLogEntry::kDataType;
+    case NT_DOUBLE:
+      return wpi::log::DoubleLogEntry::kDataType;
+    case NT_STRING:
+      return wpi::log::StringLogEntry::kDataType;
+    case NT_RAW:
+      return wpi::log::RawLogEntry::kDataType;
+    case NT_BOOLEAN_ARRAY:
+      return wpi::log::BooleanArrayLogEntry::kDataType;
+    case NT_DOUBLE_ARRAY:
+      return wpi::log::DoubleArrayLogEntry::kDataType;
+    case NT_STRING_ARRAY:
+      return wpi::log::StringArrayLogEntry::kDataType;
+    default:
+      return {};
+  }
+}
+
+void Storage::Notify(Entry* entry, unsigned int flags, bool local,
+                     std::shared_ptr<Value> value) {
+  auto& v = value ? value : entry->value;
+
+  // notifications
+  m_notifier.NotifyEntry(entry->local_id, entry->name, v,
+                         flags | (local ? NT_NOTIFY_LOCAL : 0));
+
+  if (m_dataloggers.empty()) {
+    return;
+  }
+
+  // data logging
+  // fast path the common case
+  if (entry->datalogs.empty() && (flags & NT_NOTIFY_NEW) == 0) {
+    return;
+  }
+
+  if (flags & NT_NOTIFY_DELETE) {
+    // remove all of the datalog entries
+    auto now = nt::Now();
+    for (auto&& datalog : entry->datalogs) {
+      datalog.log->Finish(datalog.entry, now);
+    }
+    entry->datalogs.clear();
+    entry->datalog_type = NT_UNASSIGNED;
+    return;
+  }
+
+  if (!v) {
+    return;
+  }
+
+  if (v->type() != entry->datalog_type) {
+    if (!entry->datalogs.empty()) {
+      // data type changed; need to finish any current logs
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->Finish(datalog.entry, v->time());
+      }
+      entry->datalogs.clear();
+    }
+
+    // create matching loggers
+    auto type = GetStorageTypeStr(v->type());
+    if (type.empty()) {
+      return;  // not a type we're going to log
+    }
+    for (auto&& logger : m_dataloggers) {
+      if (wpi::starts_with(entry->name, logger.prefix)) {
+        entry->datalogs.emplace_back(
+            logger.log,
+            logger.log->Start(
+                fmt::format("{}{}", logger.log_prefix,
+                            wpi::drop_front(entry->name, logger.prefix.size())),
+                type, "{\"source\":\"NT\"}", v->time()),
+            logger.uid);
+      }
+    }
+
+    if (entry->datalogs.empty()) {
+      return;  // we're done, nothing to log
+    }
+
+    // set datalog_type
+    entry->datalog_type = v->type();
+  }
+
+  auto time = v->time();
+  switch (v->type()) {
+    case NT_BOOLEAN: {
+      auto val = v->GetBoolean();
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->AppendBoolean(datalog.entry, val, time);
+      }
+      break;
+    }
+    case NT_DOUBLE: {
+      auto val = v->GetDouble();
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->AppendDouble(datalog.entry, val, time);
+      }
+      break;
+    }
+    case NT_STRING: {
+      auto val = v->GetString();
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->AppendString(datalog.entry, val, time);
+      }
+      break;
+    }
+    case NT_RAW: {
+      auto val = v->GetRaw();
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->AppendRaw(
+            datalog.entry,
+            {reinterpret_cast<const uint8_t*>(val.data()), val.size()}, time);
+      }
+      break;
+    }
+    case NT_BOOLEAN_ARRAY: {
+      auto val = v->GetBooleanArray();
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->AppendBooleanArray(datalog.entry, val, time);
+      }
+      break;
+    }
+    case NT_DOUBLE_ARRAY: {
+      auto val = v->GetDoubleArray();
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->AppendDoubleArray(datalog.entry, val, time);
+      }
+      break;
+    }
+    case NT_STRING_ARRAY: {
+      auto val = v->GetStringArray();
+      for (auto&& datalog : entry->datalogs) {
+        datalog.log->AppendStringArray(datalog.entry, val, time);
+      }
+      break;
+    }
+    case NT_UNASSIGNED:
+    case NT_RPC:
+      break;
+  }
+}
+
 template <typename F>
 void Storage::DeleteAllEntriesImpl(bool local, F should_delete) {
   for (auto& i : m_entries) {
     Entry* entry = i.getValue();
     if (entry->value && should_delete(entry)) {
       // notify it's being deleted
-      m_notifier.NotifyEntry(entry->local_id, i.getKey(), entry->value,
-                             NT_NOTIFY_DELETE | (local ? NT_NOTIFY_LOCAL : 0));
+      Notify(entry, NT_NOTIFY_DELETE, local);
       // remove it from idmap
       if (entry->id < m_idmap.size()) {
         m_idmap[entry->id] = nullptr;
@@ -1069,6 +1209,94 @@ unsigned int Storage::AddPolledListener(unsigned int poller,
   return uid;
 }
 
+unsigned int Storage::StartDataLog(wpi::log::DataLog& log,
+                                   std::string_view prefix,
+                                   std::string_view log_prefix) {
+  std::scoped_lock lock(m_mutex);
+
+  // create
+  unsigned int uid = m_dataloggers.emplace_back(log, prefix, log_prefix);
+  m_dataloggers[uid].uid = uid;
+
+  // start logging any matching entries
+  auto now = nt::Now();
+  for (auto&& entry : m_entries) {
+    if (!entry.second || !wpi::starts_with(entry.second->name, prefix) ||
+        !entry.second->value) {
+      continue;
+    }
+    auto type = GetStorageTypeStr(entry.second->value->type());
+    if (type.empty()) {
+      continue;  // not a type we're going to log
+    }
+    int logentry = log.Start(
+        fmt::format("{}{}", log_prefix,
+                    wpi::drop_front(entry.second->name, prefix.size())),
+        type, "{\"source\":\"NT\"}", now);
+    entry.second->datalogs.emplace_back(&log, logentry, uid);
+    // log current value
+    auto& v = *entry.second->value;
+    entry.second->datalog_type = v.type();
+    auto time = v.time();
+    switch (v.type()) {
+      case NT_BOOLEAN:
+        log.AppendBoolean(logentry, v.GetBoolean(), time);
+        break;
+      case NT_DOUBLE:
+        log.AppendDouble(logentry, v.GetDouble(), time);
+        break;
+      case NT_STRING:
+        log.AppendString(logentry, v.GetString(), time);
+        break;
+      case NT_RAW: {
+        auto val = v.GetRaw();
+        log.AppendRaw(
+            logentry,
+            {reinterpret_cast<const uint8_t*>(val.data()), val.size()}, time);
+        break;
+      }
+      case NT_BOOLEAN_ARRAY:
+        log.AppendBooleanArray(logentry, v.GetBooleanArray(), time);
+        break;
+      case NT_DOUBLE_ARRAY:
+        log.AppendDoubleArray(logentry, v.GetDoubleArray(), time);
+        break;
+      case NT_STRING_ARRAY:
+        log.AppendStringArray(logentry, v.GetStringArray(), time);
+        break;
+      default:
+        break;
+    }
+  }
+
+  return uid;
+}
+
+void Storage::StopDataLog(unsigned int uid) {
+  std::scoped_lock lock(m_mutex);
+
+  // erase the datalogger
+  auto datalogger = m_dataloggers.erase(uid);
+  if (!datalogger) {
+    return;
+  }
+
+  // finish any active entries
+  auto now = nt::Now();
+  for (auto&& entry : m_entries) {
+    if (!entry.second || entry.second->datalogs.empty()) {
+      continue;
+    }
+    auto it = std::find_if(
+        entry.second->datalogs.begin(), entry.second->datalogs.end(),
+        [&](const auto& elem) { return elem.logger_uid == uid; });
+    if (it != entry.second->datalogs.end()) {
+      it->log->Finish(it->entry, now);
+      entry.second->datalogs.erase(it);
+    }
+  }
+}
+
 bool Storage::GetPersistentEntries(
     bool periodic,
     std::vector<std::pair<std::string, std::shared_ptr<Value>>>* entries)

ntcore/src/main/native/cpp/Storage.h

@@ -18,7 +18,9 @@
 
 #include <wpi/DenseMap.h>
 #include <wpi/SmallSet.h>
+#include <wpi/SmallVector.h>
 #include <wpi/StringMap.h>
+#include <wpi/UidVector.h>
 #include <wpi/condition_variable.h>
 #include <wpi/mutex.h>
 #include <wpi/span.h>
@@ -117,6 +119,10 @@ class Storage : public IStorage {
   unsigned int AddPolledListener(unsigned int poller_uid, unsigned int local_id,
                                  unsigned int flags) const;
 
+  unsigned int StartDataLog(wpi::log::DataLog& log, std::string_view prefix,
+                            std::string_view log_prefix);
+  void StopDataLog(unsigned int uid);
+
   // Index-only
   unsigned int GetEntry(std::string_view name);
   std::vector<unsigned int> GetEntries(std::string_view prefix,
@@ -161,6 +167,29 @@ class Storage : public IStorage {
   void CancelRpcResult(unsigned int local_id, unsigned int call_uid);
 
  private:
+  struct DataLoggerEntry {
+    DataLoggerEntry(wpi::log::DataLog* log, int entry, unsigned int logger_uid)
+        : log{log}, entry{entry}, logger_uid{logger_uid} {}
+
+    wpi::log::DataLog* log;
+    int entry;
+    unsigned int logger_uid;
+  };
+
+  struct DataLogger {
+    DataLogger() = default;
+    DataLogger(wpi::log::DataLog& log, std::string_view prefix,
+               std::string_view log_prefix)
+        : log{&log}, prefix{prefix}, log_prefix{log_prefix} {}
+
+    explicit operator bool() const { return log != nullptr; }
+
+    wpi::log::DataLog* log = nullptr;
+    std::string prefix;
+    std::string log_prefix;
+    unsigned int uid;
+  };
+
   // Data for each table entry.
   struct Entry {
     explicit Entry(std::string_view name_) : name(name_) {}
@@ -195,6 +224,10 @@ class Storage : public IStorage {
     // Last UID used when calling this RPC (primarily for client use).  This
     // is incremented for each call.
     unsigned int rpc_call_uid{0};
+
+    // log entries
+    wpi::SmallVector<DataLoggerEntry, 0> datalogs;
+    NT_Type datalog_type{NT_UNASSIGNED};
   };
 
   using EntriesMap = wpi::StringMap<Entry*>;
@@ -210,6 +243,7 @@ class Storage : public IStorage {
   LocalMap m_localmap;
   RpcResultMap m_rpc_results;
   RpcBlockingCallSet m_rpc_blocking_calls;
+  wpi::UidVector<DataLogger, 4> m_dataloggers;
   // If any persistent values have changed
   mutable bool m_persistent_dirty = false;
 
@@ -255,6 +289,9 @@ class Storage : public IStorage {
   void DeleteEntryImpl(Entry* entry, std::unique_lock<wpi::mutex>& lock,
                        bool local);
 
+  void Notify(Entry* entry, unsigned int flags, bool local,
+              std::shared_ptr<Value> value = {});
+
   // Must be called with m_mutex held
   template <typename F>
   void DeleteAllEntriesImpl(bool local, F should_delete);

ntcore/src/main/native/cpp/jni/NetworkTablesJNI.cpp

@@ -12,6 +12,7 @@
 
 #include "edu_wpi_first_networktables_NetworkTablesJNI.h"
 #include "ntcore.h"
+#include "ntcore_cpp.h"
 
 using namespace wpi::java;
 
@@ -1858,6 +1859,57 @@ Java_edu_wpi_first_networktables_NetworkTablesJNI_now
   return nt::Now();
 }
 
+/*
+ * Class:     edu_wpi_first_networktables_NetworkTablesJNI
+ * Method:    startEntryDataLog
+ * Signature: (IJLjava/lang/String;Ljava/lang/String;)I
+ */
+JNIEXPORT jint JNICALL
+Java_edu_wpi_first_networktables_NetworkTablesJNI_startEntryDataLog
+  (JNIEnv* env, jclass, jint inst, jlong log, jstring prefix, jstring logPrefix)
+{
+  return nt::StartEntryDataLog(inst, *reinterpret_cast<wpi::log::DataLog*>(log),
+                               JStringRef{env, prefix},
+                               JStringRef{env, logPrefix});
+}
+
+/*
+ * Class:     edu_wpi_first_networktables_NetworkTablesJNI
+ * Method:    stopEntryDataLog
+ * Signature: (I)V
+ */
+JNIEXPORT void JNICALL
+Java_edu_wpi_first_networktables_NetworkTablesJNI_stopEntryDataLog
+  (JNIEnv*, jclass, jint logger)
+{
+  nt::StopEntryDataLog(logger);
+}
+
+/*
+ * Class:     edu_wpi_first_networktables_NetworkTablesJNI
+ * Method:    startConnectionDataLog
+ * Signature: (IJLjava/lang/String;)I
+ */
+JNIEXPORT jint JNICALL
+Java_edu_wpi_first_networktables_NetworkTablesJNI_startConnectionDataLog
+  (JNIEnv* env, jclass, jint inst, jlong log, jstring name)
+{
+  return nt::StartConnectionDataLog(
+      inst, *reinterpret_cast<wpi::log::DataLog*>(log), JStringRef{env, name});
+}
+
+/*
+ * Class:     edu_wpi_first_networktables_NetworkTablesJNI
+ * Method:    stopConnectionDataLog
+ * Signature: (I)V
+ */
+JNIEXPORT void JNICALL
+Java_edu_wpi_first_networktables_NetworkTablesJNI_stopConnectionDataLog
+  (JNIEnv*, jclass, jint logger)
+{
+  nt::StopConnectionDataLog(logger);
+}
+
 /*
  * Class:     edu_wpi_first_networktables_NetworkTablesJNI
  * Method:    createLoggerPoller

ntcore/src/main/native/cpp/ntcore_cpp.cpp

@@ -820,6 +820,57 @@ uint64_t Now() {
   return wpi::Now();
 }
 
+/*
+ * Data Logger Functions
+ */
+NT_DataLogger StartEntryDataLog(NT_Inst inst, wpi::log::DataLog& log,
+                                std::string_view prefix,
+                                std::string_view logPrefix) {
+  int i = Handle{inst}.GetTypedInst(Handle::kInstance);
+  auto ii = InstanceImpl::Get(i);
+  if (!ii) {
+    return 0;
+  }
+
+  return Handle(i, ii->storage.StartDataLog(log, prefix, logPrefix),
+                Handle::kDataLogger);
+}
+
+void StopEntryDataLog(NT_DataLogger logger) {
+  Handle handle{logger};
+  int id = handle.GetTypedIndex(Handle::kDataLogger);
+  auto ii = InstanceImpl::Get(handle.GetInst());
+  if (id < 0 || !ii) {
+    return;
+  }
+
+  ii->storage.StopDataLog(id);
+}
+
+NT_ConnectionDataLogger StartConnectionDataLog(NT_Inst inst,
+                                               wpi::log::DataLog& log,
+                                               std::string_view name) {
+  int i = Handle{inst}.GetTypedInst(Handle::kInstance);
+  auto ii = InstanceImpl::Get(i);
+  if (!ii) {
+    return 0;
+  }
+
+  return Handle(i, ii->dispatcher.StartDataLog(log, name),
+                Handle::kConnectionDataLogger);
+}
+
+void StopConnectionDataLog(NT_ConnectionDataLogger logger) {
+  Handle handle{logger};
+  int id = handle.GetTypedIndex(Handle::kConnectionDataLogger);
+  auto ii = InstanceImpl::Get(handle.GetInst());
+  if (id < 0 || !ii) {
+    return;
+  }
+
+  ii->dispatcher.StopDataLog(id);
+}
+
 /*
  * Client/Server Functions
  */

ntcore/src/main/native/include/networktables/NetworkTableInstance.h

@@ -505,6 +505,52 @@ class NetworkTableInstance final {
 
   /** @} */
 
+  /**
+   * @{
+   * @name Data Logger Functions
+   */
+
+  /**
+   * Starts logging entry changes to a DataLog.
+   *
+   * @param log data log object; lifetime must extend until StopEntryDataLog is
+   *            called or the instance is destroyed
+   * @param prefix only store entries with names that start with this prefix;
+   *               the prefix is not included in the data log entry name
+   * @param logPrefix prefix to add to data log entry names
+   * @return Data logger handle
+   */
+  NT_DataLogger StartEntryDataLog(wpi::log::DataLog& log,
+                                  std::string_view prefix,
+                                  std::string_view logPrefix);
+
+  /**
+   * Stops logging entry changes to a DataLog.
+   *
+   * @param logger data logger handle
+   */
+  static void StopEntryDataLog(NT_DataLogger logger);
+
+  /**
+   * Starts logging connection changes to a DataLog.
+   *
+   * @param log data log object; lifetime must extend until
+   *            StopConnectionDataLog is called or the instance is destroyed
+   * @param name data log entry name
+   * @return Data logger handle
+   */
+  NT_ConnectionDataLogger StartConnectionDataLog(wpi::log::DataLog& log,
+                                                 std::string_view name);
+
+  /**
+   * Stops logging connection changes to a DataLog.
+   *
+   * @param logger data logger handle
+   */
+  static void StopConnectionDataLog(NT_ConnectionDataLogger logger);
+
+  /** @} */
+
   /**
    * @{
    * @name Logger Functions

ntcore/src/main/native/include/networktables/NetworkTableInstance.inc

@@ -10,6 +10,7 @@
 #include <vector>
 
 #include "networktables/NetworkTableInstance.h"
+#include "ntcore_cpp.h"
 
 namespace nt {
 
@@ -192,6 +193,26 @@ inline const char* NetworkTableInstance::LoadEntries(
   return ::nt::LoadEntries(m_handle, filename, prefix, warn);
 }
 
+inline NT_DataLogger NetworkTableInstance::StartEntryDataLog(
+    wpi::log::DataLog& log, std::string_view prefix,
+    std::string_view logPrefix) {
+  return ::nt::StartEntryDataLog(m_handle, log, prefix, logPrefix);
+}
+
+inline void NetworkTableInstance::StopEntryDataLog(NT_DataLogger logger) {
+  ::nt::StopEntryDataLog(logger);
+}
+
+inline NT_ConnectionDataLogger NetworkTableInstance::StartConnectionDataLog(
+    wpi::log::DataLog& log, std::string_view name) {
+  return ::nt::StartConnectionDataLog(m_handle, log, name);
+}
+
+inline void NetworkTableInstance::StopConnectionDataLog(
+    NT_ConnectionDataLogger logger) {
+  ::nt::StopConnectionDataLog(logger);
+}
+
 inline NT_Logger NetworkTableInstance::AddLogger(
     std::function<void(const LogMessage& msg)> func, unsigned int min_level,
     unsigned int max_level) {

ntcore/src/main/native/include/ntcore_c.h

@@ -29,8 +29,10 @@ extern "C" {
 typedef int NT_Bool;
 
 typedef unsigned int NT_Handle;
+typedef NT_Handle NT_ConnectionDataLogger;
 typedef NT_Handle NT_ConnectionListener;
 typedef NT_Handle NT_ConnectionListenerPoller;
+typedef NT_Handle NT_DataLogger;
 typedef NT_Handle NT_Entry;
 typedef NT_Handle NT_EntryListener;
 typedef NT_Handle NT_EntryListenerPoller;

ntcore/src/main/native/include/ntcore_cpp.h

@@ -20,6 +20,10 @@
 
 #include "networktables/NetworkTableValue.h"
 
+namespace wpi::log {
+class DataLog;
+}  // namespace wpi::log
+
 /** NetworkTables (ntcore) namespace */
 namespace nt {
 
@@ -1233,6 +1237,55 @@ uint64_t Now();
 
 /** @} */
 
+/**
+ * @defgroup ntcore_data_logger_func Data Logger Functions
+ * @{
+ */
+
+/**
+ * Starts logging entry changes to a DataLog.
+ *
+ * @param inst instance handle
+ * @param log data log object; lifetime must extend until StopEntryDataLog is
+ *            called or the instance is destroyed
+ * @param prefix only store entries with names that start with this prefix;
+ *               the prefix is not included in the data log entry name
+ * @param logPrefix prefix to add to data log entry names
+ * @return Data logger handle
+ */
+NT_DataLogger StartEntryDataLog(NT_Inst inst, wpi::log::DataLog& log,
+                                std::string_view prefix,
+                                std::string_view logPrefix);
+
+/**
+ * Stops logging entry changes to a DataLog.
+ *
+ * @param logger data logger handle
+ */
+void StopEntryDataLog(NT_DataLogger logger);
+
+/**
+ * Starts logging connection changes to a DataLog.
+ *
+ * @param inst instance handle
+ * @param log data log object; lifetime must extend until StopConnectionDataLog
+ *            is called or the instance is destroyed
+ * @param name data log entry name
+ * @return Data logger handle
+ */
+NT_ConnectionDataLogger StartConnectionDataLog(NT_Inst inst,
+                                               wpi::log::DataLog& log,
+                                               std::string_view name);
+
+/**
+ * Stops logging connection changes to a DataLog.
+ *
+ * @param logger data logger handle
+ */
+void StopConnectionDataLog(NT_ConnectionDataLogger logger);
+
+/** @} */
+
 /**
  * @defgroup ntcore_logger_func Logger Functions
  * @{