T75: migrate from pmacct to ipt_NETFLOW

* Change nft to iptables in system_flow-accounting.py as ipt_NETFLOW
  is iptales plugin
* Remove specific and non-relevant pmacct options
* Add ipt_NETFLOW options
* Move 'interfaces' to 'netflow' tree
* Support more flexible 'source-address' and 'source-interface' for
  each server instead of one source
* Add migration script
* Update op mode command 'show flow-accounting'
* Update op mode command 'restart flow-accounting'

17 files changed, 749 insertions, 706 deletions

diff --git a/data/templates/ipt-netflow/ipt_NETFLOW.conf.j2 b/data/templates/ipt-netflow/ipt_NETFLOW.conf.j2
new file mode 100644
index 000000000..116b8c973
--- /dev/null
+++ b/data/templates/ipt-netflow/ipt_NETFLOW.conf.j2
@@ -0,0 +1,29 @@
+{% if netflow.engine_id is vyos_defined %}
+options ipt_NETFLOW engine-id={{ netflow.engine_id }}
+{% endif %}
+{% if netflow.max_flows is vyos_defined %}
+options ipt_NETFLOW maxflows={{ netflow.max_flows }}
+{% endif %}
+{% if netflow.sampling_rate is vyos_defined %}
+options ipt_NETFLOW sampler=random:{{ netflow.sampling_rate }}
+{% endif %}
+{% if netflow.server is vyos_defined %}
+{%     set servers_list = [] %}
+{%     for server in netflow.server %}
+{%         set d = netflow.server[server] %}
+{%         set s = (server | bracketize_ipv6) ~ ':' ~ d['port'] %}
+{%         set s = s ~ ('%' ~ d['source_interface'] if d['source_interface'] is vyos_defined else '') %}
+{%         set s = s ~ ('@' ~ d['source_address'] if d['source_address'] is vyos_defined else '') %}
+{%         set _ = servers_list.append(s) %}
+{%     endfor %}
+options ipt_NETFLOW destination={{ servers_list | join(",") }}
+{% endif %}
+{% if netflow.version is vyos_defined %}
+options ipt_NETFLOW protocol={{ netflow.version }}
+{% endif %}
+{% if netflow.inactive_timeout is vyos_defined %}
+options ipt_NETFLOW inactive_timeout={{ netflow.inactive_timeout }}
+{% endif %}
+{% if netflow.active_timeout is vyos_defined %}
+options ipt_NETFLOW active_timeout={{ netflow.active_timeout }}
+{% endif %}
diff --git a/data/templates/pmacct/override.conf.j2 b/data/templates/pmacct/override.conf.j2
deleted file mode 100644
index 44a100bb6..000000000
--- a/data/templates/pmacct/override.conf.j2
+++ /dev/null
@@ -1,17 +0,0 @@
-{% set vrf_command = 'ip vrf exec ' ~ vrf ~ ' ' if vrf is vyos_defined else '' %}
-[Unit]
-After=
-After=vyos-router.service
-ConditionPathExists=
-ConditionPathExists=/run/pmacct/uacctd.conf
-
-[Service]
-EnvironmentFile=
-ExecStart=
-ExecStart={{ vrf_command }}/usr/sbin/uacctd -f /run/pmacct/uacctd.conf
-ExecStop=/usr/libexec/vyos/system/uacctd_stop.py $MAINPID 60
-WorkingDirectory=
-WorkingDirectory=/run/pmacct
-Restart=always
-RestartSec=10
-KillMode=mixed
diff --git a/data/templates/pmacct/uacctd.conf.j2 b/data/templates/pmacct/uacctd.conf.j2
deleted file mode 100644
index d2de80df4..000000000
--- a/data/templates/pmacct/uacctd.conf.j2
+++ /dev/null
@@ -1,57 +0,0 @@
-# Genereated from VyOS configuration
-daemonize: true
-promisc: false
-syslog: daemon
-uacctd_group: 2
-uacctd_nl_size: 2097152
-snaplen: {{ packet_length }}
-aggregate: in_iface{{ ',out_iface' if enable_egress is vyos_defined }},src_mac,dst_mac,vlan,src_host,dst_host,src_port,dst_port,proto,tos,flows
-{% set pipe_size = buffer_size | int *1024 *1024 %}
-plugin_pipe_size: {{ pipe_size }}
-{# We need an integer division (//) without any remainder or fraction #}
-plugin_buffer_size: {{ pipe_size // 1000 }}
-{% if syslog_facility is vyos_defined %}
-syslog: {{ syslog_facility }}
-{% endif %}
-{% if disable_imt is not defined %}
-imt_path: /tmp/uacctd.pipe
-imt_mem_pools_number: 169
-{% endif %}
-
-{% set plugin = [] %}
-{% if netflow.server is vyos_defined %}
-{%     for server in netflow.server %}
-{%         set nf_server_key = 'nf_' ~ server | dot_colon_to_dash %}
-{%         set _ = plugin.append('nfprobe['~ nf_server_key ~ ']') %}
-{%     endfor %}
-{% endif %}
-{% if disable_imt is not defined %}
-{%     set _ = plugin.append('memory') %}
-{% endif %}
-plugins: {{ plugin | join(',') }}
-
-{% if netflow.server is vyos_defined %}
-# NetFlow servers
-{%     for server, server_config in netflow.server.items() %}
-{#         # prevent pmacct syntax error when using IPv6 flow collectors #}
-{%         set nf_server_key = 'nf_' ~ server | dot_colon_to_dash %}
-nfprobe_receiver[{{ nf_server_key }}]: {{ server | bracketize_ipv6 }}:{{ server_config.port }}
-nfprobe_version[{{ nf_server_key }}]: {{ netflow.version }}
-{%         if netflow.engine_id is vyos_defined %}
-nfprobe_engine[{{ nf_server_key }}]: {{ netflow.engine_id }}
-{%         endif %}
-{%         if netflow.max_flows is vyos_defined %}
-nfprobe_maxflows[{{ nf_server_key }}]: {{ netflow.max_flows }}
-{%         endif %}
-{%         if netflow.sampling_rate is vyos_defined %}
-sampling_rate[{{ nf_server_key }}]: {{ netflow.sampling_rate }}
-{%         endif %}
-{%         if netflow.source_address is vyos_defined %}
-nfprobe_source_ip[{{ nf_server_key }}]: {{ netflow.source_address | bracketize_ipv6 }}
-{%         endif %}
-{%         if netflow.timeout is vyos_defined %}
-nfprobe_timeouts[{{ nf_server_key }}]: expint={{ netflow.timeout.expiry_interval }}:general={{ netflow.timeout.flow_generic }}:icmp={{ netflow.timeout.icmp }}:maxlife={{ netflow.timeout.max_active_life }}:tcp.fin={{ netflow.timeout.tcp_fin }}:tcp={{ netflow.timeout.tcp_generic }}:tcp.rst={{ netflow.timeout.tcp_rst }}:udp={{ netflow.timeout.udp }}
-{%         endif %}
-
-{%     endfor %}
-{% endif %}
diff --git a/debian/control b/debian/control
index e36e20ebe..dc7b1f0a0 100644
--- a/debian/control
+++ b/debian/control
@@ -321,9 +321,6 @@ Depends:
 # For "vpn openconnect"
   ocserv,
 # End "vpn openconnect"
-# For "system flow-accounting"
-  pmacct (>= 1.6.0),
-# End "system flow-accounting"
 # For "system syslog"
   rsyslog,
 # End "system syslog"
diff --git a/interface-definitions/include/version/flow-accounting-version.xml.i b/interface-definitions/include/version/flow-accounting-version.xml.i
index 95d1e20db..3a22d391c 100644
--- a/interface-definitions/include/version/flow-accounting-version.xml.i
+++ b/interface-definitions/include/version/flow-accounting-version.xml.i
@@ -1,3 +1,3 @@
 <!-- include start from include/version/flow-accounting-version.xml.i -->
-<syntaxVersion component='flow-accounting' version='2'></syntaxVersion>
+<syntaxVersion component='flow-accounting' version='3'></syntaxVersion>
 <!-- include end -->
diff --git a/interface-definitions/system_flow-accounting.xml.in b/interface-definitions/system_flow-accounting.xml.in
index 4799205ad..53436036e 100644
--- a/interface-definitions/system_flow-accounting.xml.in
+++ b/interface-definitions/system_flow-accounting.xml.in
@@ -9,153 +9,18 @@
           <priority>990</priority>
         </properties>
         <children>
-          <leafNode name="buffer-size">
-            <properties>
-              <help>Buffer size</help>
-              <valueHelp>
-                <format>u32</format>
-                <description>Buffer size in MiB</description>
-              </valueHelp>
-              <constraint>
-                <validator name="numeric" argument="--range 0-4294967295"/>
-              </constraint>
-            </properties>
-            <defaultValue>10</defaultValue>
-          </leafNode>
-          <leafNode name="packet-length">
-            <properties>
-              <help>Specifies the maximum number of bytes to capture for each packet</help>
-              <valueHelp>
-                <format>u32:128-750</format>
-                <description>Packet length in bytes</description>
-              </valueHelp>
-              <constraint>
-                <validator name="numeric" argument="--range 128-750"/>
-              </constraint>
-            </properties>
-            <defaultValue>128</defaultValue>
-          </leafNode>
           <leafNode name="enable-egress">
             <properties>
               <help>Enable egress flow accounting</help>
               <valueless/>
             </properties>
           </leafNode>
-          <leafNode name="disable-imt">
-            <properties>
-              <help>Disable in memory table plugin</help>
-              <valueless/>
-            </properties>
-          </leafNode>
-          <leafNode name="syslog-facility">
-            <properties>
-              <help>Syslog facility for flow-accounting</help>
-              <completionHelp>
-                <list>auth authpriv cron daemon kern lpr mail mark news protocols security syslog user uucp local0 local1 local2 local3 local4 local5 local6 local7 all</list>
-              </completionHelp>
-              <valueHelp>
-                <format>auth</format>
-                <description>Authentication and authorization</description>
-              </valueHelp>
-              <valueHelp>
-                <format>authpriv</format>
-                <description>Non-system authorization</description>
-              </valueHelp>
-              <valueHelp>
-                <format>cron</format>
-                <description>Cron daemon</description>
-              </valueHelp>
-              <valueHelp>
-                <format>daemon</format>
-                <description>System daemons</description>
-              </valueHelp>
-              <valueHelp>
-                <format>kern</format>
-                <description>Kernel</description>
-              </valueHelp>
-              <valueHelp>
-                <format>lpr</format>
-                <description>Line printer spooler</description>
-              </valueHelp>
-              <valueHelp>
-                <format>mail</format>
-                <description>Mail subsystem</description>
-              </valueHelp>
-              <valueHelp>
-                <format>mark</format>
-                <description>Timestamp</description>
-              </valueHelp>
-              <valueHelp>
-                <format>news</format>
-                <description>USENET subsystem</description>
-              </valueHelp>
-              <valueHelp>
-                <format>protocols</format>
-                <description>Routing protocols (local7)</description>
-              </valueHelp>
-              <valueHelp>
-                <format>security</format>
-                <description>Authentication and authorization</description>
-              </valueHelp>
-              <valueHelp>
-                <format>syslog</format>
-                <description>Authentication and authorization</description>
-              </valueHelp>
-              <valueHelp>
-                <format>user</format>
-                <description>Application processes</description>
-              </valueHelp>
-              <valueHelp>
-                <format>uucp</format>
-                <description>UUCP subsystem</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local0</format>
-                <description>Local facility 0</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local1</format>
-                <description>Local facility 1</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local2</format>
-                <description>Local facility 2</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local3</format>
-                <description>Local facility 3</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local4</format>
-                <description>Local facility 4</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local5</format>
-                <description>Local facility 5</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local6</format>
-                <description>Local facility 6</description>
-              </valueHelp>
-              <valueHelp>
-                <format>local7</format>
-                <description>Local facility 7</description>
-              </valueHelp>
-              <valueHelp>
-                <format>all</format>
-                <description>Authentication and authorization</description>
-              </valueHelp>
-              <constraint>
-                <regex>(auth|authpriv|cron|daemon|kern|lpr|mail|mark|news|protocols|security|syslog|user|uucp|local0|local1|local2|local3|local4|local5|local6|local7|all)</regex>
-              </constraint>
-            </properties>
-          </leafNode>
-          #include <include/generic-interface-multi.xml.i>
           <node name="netflow">
             <properties>
               <help>NetFlow settings</help>
             </properties>
             <children>
+              #include <include/generic-interface-multi.xml.i>
               <leafNode name="engine-id">
                 <properties>
                   <help>NetFlow engine-id</help>
@@ -196,7 +61,6 @@
                   </constraint>
                 </properties>
               </leafNode>
-              #include <include/source-address-ipv4-ipv6.xml.i>
               <leafNode name="version">
                 <properties>
                   <help>NetFlow version to export</help>
@@ -247,119 +111,36 @@
                     </properties>
                     <defaultValue>2055</defaultValue>
                   </leafNode>
+                  #include <include/source-address-ipv4-ipv6.xml.i>
+                  #include <include/source-interface.xml.i>
                 </children>
               </tagNode>
-              <node name="timeout">
+              <leafNode name="inactive-timeout">
                 <properties>
-                  <help>NetFlow timeout values</help>
+                  <help>Flow inactivity timeout</help>
+                  <valueHelp>
+                    <format>u32:0-2147483647</format>
+                    <description>Inactive flow export timeout (seconds)</description>
+                  </valueHelp>
+                  <constraint>
+                    <validator name="numeric" argument="--range 0-2147483647"/>
+                  </constraint>
                 </properties>
-                <children>
-                  <leafNode name="expiry-interval">
-                    <properties>
-                      <help>Expiry scan interval</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>Expiry scan interval</description>
-                      </valueHelp>
-                      <constraint>
-                        <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>60</defaultValue>
-                  </leafNode>
-                  <leafNode name="flow-generic">
-                    <properties>
-                      <help>Generic flow timeout value</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>Generic flow timeout in seconds</description>
-                      </valueHelp>
-                      <constraint>
-                       <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>3600</defaultValue>
-                  </leafNode>
-                  <leafNode name="icmp">
-                    <properties>
-                      <help>ICMP timeout value</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>ICMP timeout in seconds</description>
-                      </valueHelp>
-                      <constraint>
-                       <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>300</defaultValue>
-                  </leafNode>
-                  <leafNode name="max-active-life">
-                    <properties>
-                      <help>Max active timeout value</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>Max active timeout in seconds</description>
-                      </valueHelp>
-                      <constraint>
-                       <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>604800</defaultValue>
-                  </leafNode>
-                  <leafNode name="tcp-fin">
-                    <properties>
-                      <help>TCP finish timeout value</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>TCP FIN timeout in seconds</description>
-                      </valueHelp>
-                      <constraint>
-                       <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>300</defaultValue>
-                  </leafNode>
-                  <leafNode name="tcp-generic">
-                    <properties>
-                      <help>TCP generic timeout value</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>TCP generic timeout in seconds</description>
-                      </valueHelp>
-                      <constraint>
-                       <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>3600</defaultValue>
-                  </leafNode>
-                  <leafNode name="tcp-rst">
-                    <properties>
-                      <help>TCP reset timeout value</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>TCP RST timeout in seconds</description>
-                      </valueHelp>
-                      <constraint>
-                       <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>120</defaultValue>
-                  </leafNode>
-                  <leafNode name="udp">
-                    <properties>
-                      <help>UDP timeout value</help>
-                      <valueHelp>
-                        <format>u32:0-2147483647</format>
-                        <description>UDP timeout in seconds</description>
-                      </valueHelp>
-                      <constraint>
-                       <validator name="numeric" argument="--range 0-2147483647"/>
-                      </constraint>
-                    </properties>
-                    <defaultValue>300</defaultValue>
-                  </leafNode>
-                </children>
-              </node>
+                <defaultValue>15</defaultValue>
+              </leafNode>
+              <leafNode name="active-timeout">
+                <properties>
+                  <help>Flow activity timeout</help>
+                  <valueHelp>
+                    <format>u32:0-2147483647</format>
+                    <description>Active flow export timeout (seconds)</description>
+                  </valueHelp>
+                  <constraint>
+                    <validator name="numeric" argument="--range 0-2147483647"/>
+                  </constraint>
+                </properties>
+                <defaultValue>1800</defaultValue>
+              </leafNode>
             </children>
           </node>
           #include <include/interface/vrf.xml.i>
diff --git a/op-mode-definitions/flow-accounting-op.xml.in b/op-mode-definitions/flow-accounting-op.xml.in
index 01686f0aa..00003539e 100644
--- a/op-mode-definitions/flow-accounting-op.xml.in
+++ b/op-mode-definitions/flow-accounting-op.xml.in
@@ -61,21 +61,4 @@
       </leafNode>
     </children>
   </node>
-  <node name="clear">
-    <children>
-      <node name="flow-accounting">
-        <properties>
-          <help>Clear flow accounting</help>
-        </properties>
-        <children>
-          <leafNode name="counters">
-            <properties>
-              <help>Clear flow accounting statistics</help>
-            </properties>
-            <command>${vyos_op_scripts_dir}/flow_accounting_op.py --action clear</command>
-          </leafNode>
-        </children>
-      </node>
-    </children>
-  </node>
 </interfaceDefinition>
diff --git a/op-mode-definitions/monitor-log.xml.in b/op-mode-definitions/monitor-log.xml.in
index 7139ea082..e505053b7 100644
--- a/op-mode-definitions/monitor-log.xml.in
+++ b/op-mode-definitions/monitor-log.xml.in
@@ -115,7 +115,7 @@
             <properties>
               <help>Monitor last lines of flow-accounting log</help>
             </properties>
-            <command>journalctl --no-hostname --boot --follow --unit uacctd.service</command>
+            <command>journalctl --no-hostname --boot --dmesg -g '^(ipt_NETFLOW|netflow): ' --follow</command>
           </leafNode>
           <leafNode name="frr">
             <properties>
diff --git a/op-mode-definitions/show-log.xml.in b/op-mode-definitions/show-log.xml.in
index 88d69a27c..be30522a0 100755
--- a/op-mode-definitions/show-log.xml.in
+++ b/op-mode-definitions/show-log.xml.in
@@ -567,7 +567,7 @@
             <properties>
               <help>Show log for flow-accounting</help>
             </properties>
-            <command>journalctl --no-hostname --boot --unit uacctd.service</command>
+            <command>journalctl --no-hostname --boot --dmesg -g '^(ipt_NETFLOW|netflow): '</command>
           </leafNode>
           <leafNode name="frr">
             <properties>
diff --git a/python/vyos/ipt_netflow.py b/python/vyos/ipt_netflow.py
new file mode 100644
index 000000000..b1630f30c
--- /dev/null
+++ b/python/vyos/ipt_netflow.py
@@ -0,0 +1,174 @@
+# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library.  If not, see <http://www.gnu.org/licenses/>.
+
+# Package to stop/start ipt_NETFLOW kernel module
+
+# Provides functions stop(), start() and set_watched_iptables_interfaces()
+
+from vyos.utils.kernel import check_kmod
+from vyos.utils.kernel import unload_kmod
+from vyos.utils.process import cmd
+from vyos import ConfigError
+
+module_name = 'ipt_NETFLOW'
+iptables_ingress_netflow_table = 'raw'
+iptables_ingress_netflow_chain = 'PREROUTING'
+iptables_egress_netflow_table = 'mangle'
+iptables_egress_netflow_chain = 'POSTROUTING'
+
+
+# get iptables rule dict for chain in table
+def _iptables_get_rules(command, chain, table):
+    # define list with rules
+    rules = []
+
+    # run iptables, save output and split it by lines
+    iptables_command = f'{command} -vn -t {table} -L {chain}'
+    tmp = cmd(iptables_command, message='Failed to get flows list')
+    lines = tmp.splitlines()
+
+    # Sample output to parse:
+    #   vyos@vyos:~$ sudo iptables -vn -t raw -L PREROUTING
+    #   Chain PREROUTING (policy ACCEPT 0 packets, 0 bytes)
+    #   pkts bytes target     prot opt in     out     source               destination
+    #      0     0 NETFLOW    0    --  eth0   *       0.0.0.0/0            0.0.0.0/0           NETFLOW
+
+    # Check that format is as expected
+    if len(lines) < 2:
+        raise ConfigError(f'Unexpected output from {command}, too few lines')
+    if not lines[0].startswith(f'Chain {chain}'):
+        raise ConfigError(f'Unexpected first line in output of {command}: "{lines[0]}"')
+    columns = lines[1].split()
+
+    # parse each line and add information to list
+    rulenum = 0
+    for current_rule in lines[2:]:
+        rulenum += 1
+        current_rule_parsed = current_rule.split()
+        current_rule_parsed = {
+            columns[i]: current_rule_parsed[i]
+            for i in range(min(len(current_rule_parsed), len(columns)))
+        }
+        if current_rule_parsed.get('target', '') != 'NETFLOW':
+            continue
+
+        rules.append(
+            {
+                'interface-in': current_rule_parsed.get("in", ''),
+                'interface-out': current_rule_parsed.get("out", ''),
+                'table': table,
+                'rulenum': rulenum,
+            }
+        )
+
+    # return list with rules
+    return rules
+
+
+def _iptables_config(command, configured_ifaces, direction):
+    # define list of nftables commands to modify settings
+    iptables_commands = []
+
+    if direction == "ingress":
+        iptables_table = iptables_ingress_netflow_table
+        iptables_chain = iptables_ingress_netflow_chain
+    elif direction == "egress":
+        iptables_table = iptables_egress_netflow_table
+        iptables_chain = iptables_egress_netflow_chain
+    else:
+        raise ConfigError(f'_iptables_config: Unexpected direction="{direction}"')
+
+    # prepare extended list with configured interfaces
+    configured_ifaces_extended = []
+    for iface in configured_ifaces:
+        configured_ifaces_extended.append({'iface': iface})
+
+    # get currently configured interfaces with iptables rules
+    active_rules = _iptables_get_rules(command, iptables_chain, iptables_table)
+
+    # compare current active list with configured one and delete excessive interfaces, add missed
+    active_ifaces = []
+    interface_key = 'interface-out' if direction == "egress" else "interface-in"
+    rulenums_delete = []
+    for rule in active_rules:
+        interface = rule[interface_key]
+        if interface not in configured_ifaces:
+            rulenums_delete.append(rule['rulenum'])
+        else:
+            active_ifaces.append({'iface': interface})
+
+    # It is important to delete rule with bigger rulenum first, so that other
+    # rulenums are not changed
+    rulenums_delete.sort(reverse=True)
+    for rulenum in rulenums_delete:
+        iptables_commands.append(
+            f'{command} -t {iptables_table} -D {iptables_chain} {rulenum}'
+        )
+
+    # do not create new rules for already configured interfaces
+    for iface in active_ifaces:
+        if iface in configured_ifaces_extended:
+            configured_ifaces_extended.remove(iface)
+
+    # create missed rules
+    for iface_extended in configured_ifaces_extended:
+        iface = iface_extended['iface']
+        iface_option = "o" if direction == "egress" else "i"
+        # iptables -t raw -A PREROUTING -j NETFLOW -i eth0
+        rule_definition = f'{command} -t {iptables_table} -A {iptables_chain} -j NETFLOW -{iface_option} {iface}'
+        iptables_commands.append(rule_definition)
+
+    # change iptables
+    for command in iptables_commands:
+        cmd(command, raising=ConfigError)
+
+
+def _iptables_config_v4_and_v6(configured_ifaces, direction):
+    for command in 'iptables', 'ip6tables':
+        _iptables_config(command, configured_ifaces, direction)
+
+
+def set_watched_iptables_interfaces(ingress_interfaces, egress_interfaces):
+    """
+    Update iptables and ip6tables rules so that ipt_NETFLOW watches
+    exact list of interfaces in ingress_interfaces for ingress table/chain
+    and egress_interfaces for egress table/chain
+    """
+    _iptables_config_v4_and_v6(ingress_interfaces, 'ingress')
+    _iptables_config_v4_and_v6(egress_interfaces, 'egress')
+
+
+def stop():
+    """
+    Stop ipt_NETFLOW: remove all iptables rules that use it
+    and remove module
+    """
+    set_watched_iptables_interfaces([], [])
+
+    unload_kmod(module_name)
+
+
+def start(ingress_interfaces, egress_interfaces):
+    """
+    Start ipt_NETFLOW:
+
+    * Load ipt_NETFLOW kernel module
+    * Install iptables and ip6tables rules for
+      ingress_interfaces and egress_interfaces
+    """
+
+    check_kmod(module_name)
+
+    set_watched_iptables_interfaces(ingress_interfaces, egress_interfaces)
diff --git a/python/vyos/utils/kernel.py b/python/vyos/utils/kernel.py
index 4d8544670..4255d88d5 100644
--- a/python/vyos/utils/kernel.py
+++ b/python/vyos/utils/kernel.py
@@ -30,6 +30,11 @@ def check_kmod(k_mod):
             if call(f'modprobe {module}') != 0:
                 raise ConfigError(f'Loading Kernel module {module} failed')
 
+
+def is_module_loaded(module):
+    """Common utility function to check whether module is loaded"""
+    return os.path.exists(f'/sys/module/{module}')
+
 def unload_kmod(k_mod):
     """ Common utility function to unload required kernel modules on demand """
     from vyos import ConfigError
@@ -37,7 +42,7 @@ def unload_kmod(k_mod):
     if isinstance(k_mod, str):
         k_mod = k_mod.split()
     for module in k_mod:
-        if os.path.exists(f'/sys/module/{module}'):
+        if is_module_loaded(module):
             if call(f'rmmod {module}') != 0:
                 raise ConfigError(f'Unloading Kernel module {module} failed')
 
diff --git a/smoketest/config-tests/bgp-azure-ipsec-gateway b/smoketest/config-tests/bgp-azure-ipsec-gateway
index 0d683c921..bef48fd6d 100644
--- a/smoketest/config-tests/bgp-azure-ipsec-gateway
+++ b/smoketest/config-tests/bgp-azure-ipsec-gateway
@@ -111,17 +111,16 @@ set system conntrack modules sqlnet
 set system conntrack modules tftp
 set system console device ttyS0 speed '115200'
 set system domain-name 'vyos.net'
-set system flow-accounting interface 'eth1'
-set system flow-accounting interface 'vti31'
-set system flow-accounting interface 'vti32'
-set system flow-accounting interface 'vti41'
-set system flow-accounting interface 'vti42'
-set system flow-accounting interface 'vti51'
-set system flow-accounting interface 'vti52'
+set system flow-accounting netflow interface 'eth1'
+set system flow-accounting netflow interface 'vti31'
+set system flow-accounting netflow interface 'vti32'
+set system flow-accounting netflow interface 'vti41'
+set system flow-accounting netflow interface 'vti42'
+set system flow-accounting netflow interface 'vti51'
+set system flow-accounting netflow interface 'vti52'
 set system flow-accounting netflow server 10.0.1.1 port '2055'
-set system flow-accounting netflow source-address '192.168.34.37'
+set system flow-accounting netflow server 10.0.1.1 source-address '192.168.34.37'
 set system flow-accounting netflow version '10'
-set system flow-accounting syslog-facility 'daemon'
 set system host-name 'azure-gw-01'
 set system login radius server 192.0.2.253 key 'secret1234'
 set system login radius server 192.0.2.253 port '1812'
diff --git a/smoketest/config-tests/bgp-big-as-cloud b/smoketest/config-tests/bgp-big-as-cloud
index f71a51be3..072f5258a 100644
--- a/smoketest/config-tests/bgp-big-as-cloud
+++ b/smoketest/config-tests/bgp-big-as-cloud
@@ -830,13 +830,11 @@ set system conntrack modules sip
 set system conntrack modules sqlnet
 set system conntrack modules tftp
 set system console device ttyS0 speed '115200'
-set system flow-accounting disable-imt
-set system flow-accounting interface 'eth0.4088'
-set system flow-accounting interface 'eth0.4089'
+set system flow-accounting netflow interface 'eth0.4088'
+set system flow-accounting netflow interface 'eth0.4089'
 set system flow-accounting netflow engine-id '1'
 set system flow-accounting netflow server 192.0.2.55 port '2055'
 set system flow-accounting netflow version '9'
-set system flow-accounting syslog-facility 'daemon'
 set system host-name 'vyos'
 set system login user vyos authentication encrypted-password '$6$2Ta6TWHd/U$NmrX0x9kexCimeOcYK1MfhMpITF9ELxHcaBU/znBq.X2ukQOj61fVI2UYP/xBzP4QtiTcdkgs7WOQMHWsRymO/'
 set system login user vyos authentication plaintext-password ''
diff --git a/smoketest/scripts/cli/test_system_flow-accounting.py b/smoketest/scripts/cli/test_system_flow-accounting.py
index 7778c393f..80bcd6b89 100755
--- a/smoketest/scripts/cli/test_system_flow-accounting.py
+++ b/smoketest/scripts/cli/test_system_flow-accounting.py
@@ -20,17 +20,53 @@ from base_vyostest_shim import VyOSUnitTestSHIM
 
 from vyos.configsession import ConfigSessionError
 from vyos.ifconfig import Section
-from vyos.template import bracketize_ipv6
+from vyos.utils.kernel import is_module_loaded
+from vyos.utils.kernel import get_module_data
 from vyos.utils.process import cmd
-from vyos.utils.process import process_named_running
-from vyos.utils.file import read_file
 
-PROCESS_NAME = 'uacctd'
+module_name = 'ipt_NETFLOW'
 base_path = ['system', 'flow-accounting']
 
-uacctd_conf = '/run/pmacct/uacctd.conf'
 
 class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
+
+    def _get_iptables_watched_interfaces(self, command, table, chain, column_name):
+        iptables_command = f'{command} -vn -t {table} -L {chain}'
+        data = cmd(iptables_command, message='Failed to get flows list')
+        data = data.splitlines()
+        self.assertGreaterEqual(
+            len(data), 2, "Unexpected output of {command}, should be at least two lines"
+        )
+        column_index = data[1].split().index(column_name)
+        interfaces = [
+            line.split()[column_index] for line in data[2:] if 'NETFLOW' in line
+        ]
+        return interfaces
+
+    def _get_iptables_watched_ingress_interfaces(self, command):
+        return self._get_iptables_watched_interfaces(command, 'raw', 'PREROUTING', 'in')
+
+    def _get_iptables_watched_egress_interfaces(self, command):
+        return self._get_iptables_watched_interfaces(
+            command, 'mangle', 'POSTROUTING', 'out'
+        )
+
+    def _assert_ingress_interfaces(self, interfaces):
+        for command in 'iptables', 'ip6tables':
+            self.assertEqual(
+                set(self._get_iptables_watched_ingress_interfaces(command)),
+                set(interfaces),
+                command,
+            )
+
+    def _assert_egress_interfaces(self, interfaces):
+        for command in 'iptables', 'ip6tables':
+            self.assertEqual(
+                set(self._get_iptables_watched_egress_interfaces(command)),
+                set(interfaces),
+                command,
+            )
+
     @classmethod
     def setUpClass(cls):
         super(TestSystemFlowAccounting, cls).setUpClass()
@@ -41,106 +77,83 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
 
     def tearDown(self):
         # after service removal process must no longer run
-        self.assertTrue(process_named_running(PROCESS_NAME))
+        self.assertTrue(is_module_loaded(module_name))
 
         self.cli_delete(base_path)
         self.cli_commit()
 
         # after service removal process must no longer run
-        self.assertFalse(process_named_running(PROCESS_NAME))
+        self.assertFalse(is_module_loaded(module_name))
+        self._assert_ingress_interfaces([])
+        self._assert_egress_interfaces([])
 
     def test_basic(self):
-        buffer_size = '5' # MiB
-        syslog = 'all'
-
-        self.cli_set(base_path + ['buffer-size', buffer_size])
-        self.cli_set(base_path + ['syslog-facility', syslog])
+        engine_id = '33'
+        self.cli_set(base_path + ['netflow', 'engine-id', engine_id])
 
         # You need to configure at least one interface for flow-accounting
         with self.assertRaises(ConfigSessionError):
             self.cli_commit()
         for interface in Section.interfaces('ethernet'):
-            self.cli_set(base_path + ['interface', interface])
+            self.cli_set(base_path + ['netflow', 'interface', interface])
 
-        # commit changes
+        # You need to configure at least one NetFlow server
+        with self.assertRaises(ConfigSessionError):
+            self.cli_commit()
+
+        netflow_server = '11.22.33.44'
+        self.cli_set(base_path + ['netflow', 'server', netflow_server])
+
+        # commit changes, this time should work
         self.cli_commit()
 
         # verify configuration
-        nftables_output = cmd('sudo nft list chain raw VYOS_PREROUTING_HOOK').splitlines()
-        for interface in Section.interfaces('ethernet'):
-            rule_found = False
-            ifname_search = f'iifname "{interface}"'
-
-            for nftables_line in nftables_output:
-                if 'FLOW_ACCOUNTING_RULE' in nftables_line and ifname_search in nftables_line:
-                    self.assertIn('group 2', nftables_line)
-                    self.assertIn('snaplen 128', nftables_line)
-                    self.assertIn('queue-threshold 100', nftables_line)
-                    rule_found = True
-                    break
-
-            self.assertTrue(rule_found)
-
-        uacctd = read_file(uacctd_conf)
-        # circular queue size - buffer_size
-        tmp = int(buffer_size) *1024 *1024
-        self.assertIn(f'plugin_pipe_size: {tmp}', uacctd)
-        # transfer buffer size - recommended value from pmacct developers 1/1000 of pipe size
-        tmp = int(buffer_size) *1024 *1024
-        # do an integer division
-        tmp //= 1000
-        self.assertIn(f'plugin_buffer_size: {tmp}', uacctd)
-
-        # when 'disable-imt' is not configured on the CLI it must be present
-        self.assertIn(f'imt_path: /tmp/uacctd.pipe', uacctd)
-        self.assertIn(f'imt_mem_pools_number: 169', uacctd)
-        self.assertIn(f'syslog: {syslog}', uacctd)
-        self.assertIn(f'plugins: memory', uacctd)
+        self._assert_ingress_interfaces(Section.interfaces('ethernet'))
+        self._assert_egress_interfaces([])
+
+        module_data = get_module_data(module_name)
+        self.assertEqual(engine_id, module_data['parameters']['engine_id'])
+
 
     def test_netflow(self):
         engine_id = '33'
         max_flows = '667'
-        sampling_rate = '100'
-        source_address = '192.0.2.1'
         dummy_if = 'dum3842'
         agent_address = '192.0.2.10'
         version = '10'
-        tmo_expiry = '120'
-        tmo_flow = '1200'
-        tmo_icmp = '60'
-        tmo_max = '50000'
-        tmo_tcp_fin = '100'
-        tmo_tcp_generic = '120'
-        tmo_tcp_rst = '99'
-        tmo_udp = '10'
+        active_timeout = '900'
+        inactive_timeout = '30'
 
+        source_ipv4_address = '192.0.2.1'
+        source_ipv6_address = '2001:db8::ab'
         netflow_server = {
-            '11.22.33.44' : { },
-            '55.66.77.88' : { 'port' : '6000' },
-            '2001:db8::1' : { },
+            '11.22.33.44': {},
+            '55.66.77.88': {'port': '6000'},
+            '100.12.14.1': {'source-interface': dummy_if},
+            '203.0.113.21': {'port': '3000', 'source-address': source_ipv4_address},
+            '2001:db8::1': {'source-address': source_ipv6_address},
         }
-
-        self.cli_set(['interfaces', 'dummy', dummy_if, 'address', agent_address + '/32'])
-        self.cli_set(['interfaces', 'dummy', dummy_if, 'address', source_address + '/32'])
+        # ipt_NETFLOW sorts destinations by IP
+        expected_destination = '11.22.33.44:2055,55.66.77.88:6000,100.12.14.1:2055%dum3842,203.0.113.21:3000@192.0.2.1,[2001:db8::1]:2055@2001:db8::ab'
+
+        self.cli_set(
+            ['interfaces', 'dummy', dummy_if, 'address', agent_address + '/32']
+        )
+        self.cli_set(
+            ['interfaces', 'dummy', dummy_if, 'address', source_ipv4_address + '/32']
+        )
+        self.cli_set(
+            ['interfaces', 'dummy', dummy_if, 'address', source_ipv6_address + '/128']
+        )
 
         for interface in Section.interfaces('ethernet'):
-            self.cli_set(base_path + ['interface', interface])
+            self.cli_set(base_path + ['netflow', 'interface', interface])
 
         self.cli_set(base_path + ['netflow', 'engine-id', engine_id])
         self.cli_set(base_path + ['netflow', 'max-flows', max_flows])
-        self.cli_set(base_path + ['netflow', 'sampling-rate', sampling_rate])
-        self.cli_set(base_path + ['netflow', 'source-address', source_address])
         self.cli_set(base_path + ['netflow', 'version', version])
-
-        # timeouts
-        self.cli_set(base_path + ['netflow', 'timeout', 'expiry-interval', tmo_expiry])
-        self.cli_set(base_path + ['netflow', 'timeout', 'flow-generic', tmo_flow])
-        self.cli_set(base_path + ['netflow', 'timeout', 'icmp', tmo_icmp])
-        self.cli_set(base_path + ['netflow', 'timeout', 'max-active-life', tmo_max])
-        self.cli_set(base_path + ['netflow', 'timeout', 'tcp-fin', tmo_tcp_fin])
-        self.cli_set(base_path + ['netflow', 'timeout', 'tcp-generic', tmo_tcp_generic])
-        self.cli_set(base_path + ['netflow', 'timeout', 'tcp-rst', tmo_tcp_rst])
-        self.cli_set(base_path + ['netflow', 'timeout', 'udp', tmo_udp])
+        self.cli_set(base_path + ['netflow', 'active-timeout', active_timeout])
+        self.cli_set(base_path + ['netflow', 'inactive-timeout', inactive_timeout])
 
         # You need to configure at least one netflow server
         with self.assertRaises(ConfigSessionError):
@@ -150,41 +163,120 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
             self.cli_set(base_path + ['netflow', 'server', server])
             if 'port' in server_config:
                 self.cli_set(base_path + ['netflow', 'server', server, 'port', server_config['port']])
+            if 'source-address' in server_config:
+                self.cli_set(
+                    base_path
+                    + [
+                        'netflow',
+                        'server',
+                        server,
+                        'source-address',
+                        server_config['source-address'],
+                    ]
+                )
+            if 'source-interface' in server_config:
+                self.cli_set(
+                    base_path
+                    + [
+                        'netflow',
+                        'server',
+                        server,
+                        'source-interface',
+                        server_config['source-interface'],
+                    ]
+                )
 
         # commit changes
         self.cli_commit()
 
-        uacctd = read_file(uacctd_conf)
+        module_data = get_module_data(module_name)
 
-        tmp = []
-        for server, server_config in netflow_server.items():
-            tmp_srv = server
-            tmp_srv = tmp_srv.replace('.', '-')
-            tmp_srv = tmp_srv.replace(':', '-')
-            tmp.append(f'nfprobe[nf_{tmp_srv}]')
-        tmp.append('memory')
-        self.assertIn('plugins: ' + ','.join(tmp), uacctd)
+        self.assertEqual(engine_id, module_data['parameters']['engine_id'])
+        self.assertEqual(max_flows, module_data['parameters']['maxflows'])
+        self.assertEqual(expected_destination, module_data['parameters']['destination'])
+        self.assertEqual(version, module_data['parameters']['protocol'])
+        self.assertEqual(active_timeout, module_data['parameters']['active_timeout'])
+        self.assertEqual(
+            inactive_timeout, module_data['parameters']['inactive_timeout']
+        )
 
-        for server, server_config in netflow_server.items():
-            tmp_srv = server
-            tmp_srv = tmp_srv.replace('.', '-')
-            tmp_srv = tmp_srv.replace(':', '-')
-
-            self.assertIn(f'nfprobe_engine[nf_{tmp_srv}]: {engine_id}', uacctd)
-            self.assertIn(f'nfprobe_maxflows[nf_{tmp_srv}]: {max_flows}', uacctd)
-            self.assertIn(f'sampling_rate[nf_{tmp_srv}]: {sampling_rate}', uacctd)
-            self.assertIn(f'nfprobe_source_ip[nf_{tmp_srv}]: {source_address}', uacctd)
-            self.assertIn(f'nfprobe_version[nf_{tmp_srv}]: {version}', uacctd)
+        # Test module reload with new parameters
+        engine_id = '73'
+        self.cli_set(base_path + ['netflow', 'engine-id', engine_id])
+        self.cli_commit()
 
-            if 'port' in server_config:
-                self.assertIn(f'nfprobe_receiver[nf_{tmp_srv}]: {bracketize_ipv6(server)}', uacctd)
-            else:
-                self.assertIn(f'nfprobe_receiver[nf_{tmp_srv}]: {bracketize_ipv6(server)}:2055', uacctd)
+        module_data = get_module_data(module_name)
 
-            self.assertIn(f'nfprobe_timeouts[nf_{tmp_srv}]: expint={tmo_expiry}:general={tmo_flow}:icmp={tmo_icmp}:maxlife={tmo_max}:tcp.fin={tmo_tcp_fin}:tcp={tmo_tcp_generic}:tcp.rst={tmo_tcp_rst}:udp={tmo_udp}', uacctd)
+        self.assertEqual(engine_id, module_data['parameters']['engine_id'])
 
         self.cli_delete(['interfaces', 'dummy', dummy_if])
 
 
+    def test_iptables(self):
+        netflow_server = '11.22.33.44'
+        self.cli_set(base_path + ['netflow', 'server', netflow_server])
+
+        dummy_ifs = [
+            'dum4000',
+            'dum4001',
+            'dum4002',
+            'dum4003',
+        ]
+
+        self.cli_set(['interfaces', 'dummy', dummy_ifs[0], 'address', '192.0.2.100/32'])
+        self.cli_set(['interfaces', 'dummy', dummy_ifs[1], 'address', '192.0.2.101/32'])
+        self.cli_set(['interfaces', 'dummy', dummy_ifs[2], 'address', '192.0.2.102/32'])
+        self.cli_set(['interfaces', 'dummy', dummy_ifs[3], 'address', '192.0.2.103/32'])
+
+        # * three interfaces
+        for i in range(3):
+            self.cli_set(base_path + ['netflow', 'interface', dummy_ifs[i]])
+        self.cli_commit()
+        self._assert_ingress_interfaces(dummy_ifs[0:3])
+        self._assert_egress_interfaces([])
+
+        # * Then delete one
+        self.cli_delete(base_path + ['netflow', 'interface', dummy_ifs[1]])
+        self.cli_commit()
+        self._assert_ingress_interfaces([dummy_ifs[0], dummy_ifs[2]])
+        self._assert_egress_interfaces([])
+
+        # * Then add one
+        self.cli_set(base_path + ['netflow', 'interface', dummy_ifs[3]])
+        self.cli_commit()
+        self._assert_ingress_interfaces([dummy_ifs[0], dummy_ifs[2], dummy_ifs[3]])
+        self._assert_egress_interfaces([])
+
+        # * enable egress
+        self.cli_set(base_path + ['enable-egress'])
+        self.cli_commit()
+        self._assert_ingress_interfaces([dummy_ifs[0], dummy_ifs[2], dummy_ifs[3]])
+        self._assert_egress_interfaces([dummy_ifs[0], dummy_ifs[2], dummy_ifs[3]])
+
+    def test_sampler(self):
+        # Separate test because if --enable-sampler is not given to configure of
+        # ipt_NETFLOW this parameter is not available
+        sampling_rate = '100'
+        self.cli_set(base_path + ['netflow', 'sampling-rate', sampling_rate])
+
+        for interface in Section.interfaces('ethernet'):
+            self.cli_set(base_path + ['netflow', 'interface', interface])
+
+        netflow_server = '11.22.33.44'
+        self.cli_set(base_path + ['netflow', 'server', netflow_server])
+
+        # commit changes, this time should work
+        self.cli_commit()
+
+        module_data = get_module_data(module_name)
+
+        if 'sampler' not in module_data['parameters']:
+            self.skipTest("ipt_NETFLOW has no sampler parameter")
+
+        self.assertEqual(
+            f'random:{sampling_rate}', module_data['parameters']['sampler']
+        )
+
+
 if __name__ == '__main__':
     unittest.main(verbosity=2)
diff --git a/src/conf_mode/system_flow-accounting.py b/src/conf_mode/system_flow-accounting.py
index 618227fc0..b23ee77fd 100755
--- a/src/conf_mode/system_flow-accounting.py
+++ b/src/conf_mode/system_flow-accounting.py
@@ -17,6 +17,7 @@
 import os
 import re
 
+from ipaddress import ip_interface
 from sys import exit
 
 from vyos.config import Config
@@ -24,119 +25,19 @@ from vyos.config import config_dict_merge
 from vyos.configverify import verify_vrf
 from vyos.configverify import verify_interface_exists
 from vyos.template import render
-from vyos.utils.process import call
-from vyos.utils.process import cmd
-from vyos.utils.process import run
+from vyos.utils.file import read_file
 from vyos.utils.network import is_addr_assigned
 from vyos import ConfigError
 from vyos import airbag
+from vyos import ipt_netflow
 airbag.enable()
 
-uacctd_conf_path = '/run/pmacct/uacctd.conf'
-systemd_service = 'uacctd.service'
-systemd_override = f'/run/systemd/system/{systemd_service}.d/override.conf'
-nftables_nflog_table = 'raw'
-nftables_nflog_chain = 'VYOS_PREROUTING_HOOK'
-egress_nftables_nflog_table = 'inet mangle'
-egress_nftables_nflog_chain = 'FORWARD'
-
-# get nftables rule dict for chain in table
-def _nftables_get_nflog(chain, table):
-    # define list with rules
-    rules = []
-
-    # prepare regex for parsing rules
-    rule_pattern = '[io]ifname "(?P<interface>[\w\.\*\-]+)".*handle (?P<handle>[\d]+)'
-    rule_re = re.compile(rule_pattern)
-
-    # run nftables, save output and split it by lines
-    nftables_command = f'nft -a list chain {table} {chain}'
-    tmp = cmd(nftables_command, message='Failed to get flows list')
-    # parse each line and add information to list
-    for current_rule in tmp.splitlines():
-        if 'FLOW_ACCOUNTING_RULE' not in current_rule:
-            continue
-        current_rule_parsed = rule_re.search(current_rule)
-        if current_rule_parsed:
-            groups = current_rule_parsed.groupdict()
-            rules.append({ 'interface': groups["interface"], 'table': table, 'handle': groups["handle"] })
-
-    # return list with rules
-    return rules
-
-def _nftables_config(configured_ifaces, direction, length=None):
-    # define list of nftables commands to modify settings
-    nftable_commands = []
-    nftables_chain = nftables_nflog_chain
-    nftables_table = nftables_nflog_table
-
-    if direction == "egress":
-        nftables_chain = egress_nftables_nflog_chain
-        nftables_table = egress_nftables_nflog_table
-
-    # prepare extended list with configured interfaces
-    configured_ifaces_extended = []
-    for iface in configured_ifaces:
-        configured_ifaces_extended.append({ 'iface': iface })
-
-    # get currently configured interfaces with nftables rules
-    active_nflog_rules = _nftables_get_nflog(nftables_chain, nftables_table)
-
-    # compare current active list with configured one and delete excessive interfaces, add missed
-    active_nflog_ifaces = []
-    for rule in active_nflog_rules:
-        interface = rule['interface']
-        if interface not in configured_ifaces:
-            table = rule['table']
-            handle = rule['handle']
-            nftable_commands.append(f'nft delete rule {table} {nftables_chain} handle {handle}')
-        else:
-            active_nflog_ifaces.append({
-                'iface': interface,
-            })
-
-    # do not create new rules for already configured interfaces
-    for iface in active_nflog_ifaces:
-        if iface in active_nflog_ifaces and iface in configured_ifaces_extended:
-            configured_ifaces_extended.remove(iface)
-
-    # create missed rules
-    for iface_extended in configured_ifaces_extended:
-        iface = iface_extended['iface']
-        iface_prefix = "o" if direction == "egress" else "i"
-        rule_definition = f'{iface_prefix}ifname "{iface}" counter log group 2 snaplen {length} queue-threshold 100 comment "FLOW_ACCOUNTING_RULE"'
-        nftable_commands.append(f'nft insert rule {nftables_table} {nftables_chain} {rule_definition}')
-        # Also add IPv6 ingres logging
-        if nftables_table == nftables_nflog_table:
-            nftable_commands.append(f'nft insert rule ip6 {nftables_table} {nftables_chain} {rule_definition}')
-
-    # change nftables
-    for command in nftable_commands:
-        cmd(command, raising=ConfigError)
-
-
-def _nftables_trigger_setup(operation: str) -> None:
-    """Add a dummy rule to unlock the main pmacct loop with a packet-trigger
-
-    Args:
-        operation (str): 'add' or 'delete' a trigger
-    """
-    # check if a chain exists
-    table_exists = False
-    if run('nft -snj list table ip pmacct') == 0:
-        table_exists = True
-
-    if operation == 'delete' and table_exists:
-        nft_cmd: str = 'nft delete table ip pmacct'
-        cmd(nft_cmd, raising=ConfigError)
-    if operation == 'add' and not table_exists:
-        nft_cmds: list[str] = [
-            'nft add table ip pmacct',
-            'nft add chain ip pmacct pmacct_out { type filter hook output priority raw - 50 \\; policy accept \\; }',
-            'nft add rule ip pmacct pmacct_out oif lo ip daddr 127.0.254.0 counter log group 2 snaplen 1 queue-threshold 0 comment NFLOG_TRIGGER'
-        ]
-        for nft_cmd in nft_cmds:
-            cmd(nft_cmd, raising=ConfigError)
+ipt_netflow_conf_path = '/etc/modprobe.d/ipt_NETFLOW.conf'
+
+# Variable to store between generate and apply
+# whether module configuration was changed
+# and module reload is needed
+need_reload = True
 
 
 def get_config(config=None):
@@ -166,104 +67,128 @@ def get_config(config=None):
 
     return flow_accounting
 
+
 def verify(flow_config):
     if not flow_config:
         return None
 
-    # check if collector is enabled
-    if 'netflow' not in flow_config and 'disable_imt' in flow_config:
-        raise ConfigError('You need to configure NetFlow, ' \
-                          'or not set "disable-imt" for flow-accounting!')
-
     # Check if at least one interface is configured
-    if 'interface' not in flow_config:
+    if 'netflow' not in flow_config or 'interface' not in flow_config['netflow']:
         raise ConfigError('Flow accounting requires at least one interface to ' \
                           'be configured!')
 
     # check that all configured interfaces exists in the system
-    for interface in flow_config['interface']:
+    for interface in flow_config['netflow']['interface']:
         verify_interface_exists(flow_config, interface, warning_only=True)
 
+    # check if at least one NetFlow collector is configured
+    if 'server' not in flow_config['netflow']:
+        raise ConfigError('You need to configure at least one NetFlow server!')
     verify_vrf(flow_config)
 
-    # check NetFlow configuration
-    if 'netflow' in flow_config:
-        # check if vrf is defined for netflow
-        netflow_vrf = None
-        if 'vrf' in flow_config:
-            netflow_vrf = flow_config['vrf']
-
-        # check if at least one NetFlow collector is configured if NetFlow configuration is presented
-        if 'server' not in flow_config['netflow']:
-            raise ConfigError('You need to configure at least one NetFlow server!')
-
-        # Check if configured netflow source-address exist in the system
-        if 'source_address' in flow_config['netflow']:
-            if not is_addr_assigned(flow_config['netflow']['source_address'], netflow_vrf):
-                tmp = flow_config['netflow']['source_address']
-                raise ConfigError(f'Configured "netflow source-address {tmp}" does not exist on the system!')
-
-        # Check if engine-id compatible with selected protocol version
-        if 'engine_id' in flow_config['netflow']:
-            v5_filter = '^(\d|[1-9]\d|1\d{2}|2[0-4]\d|25[0-5]):(\d|[1-9]\d|1\d{2}|2[0-4]\d|25[0-5])$'
-            v9v10_filter = '^(\d|[1-9]\d{1,8}|[1-3]\d{9}|4[01]\d{8}|42[0-8]\d{7}|429[0-3]\d{6}|4294[0-8]\d{5}|42949[0-5]\d{4}|429496[0-6]\d{3}|4294967[01]\d{2}|42949672[0-8]\d|429496729[0-5])$'
-            engine_id = flow_config['netflow']['engine_id']
-            version = flow_config['netflow']['version']
-
-            if flow_config['netflow']['version'] == '5':
-                regex_filter = re.compile(v5_filter)
-                if not regex_filter.search(engine_id):
-                    raise ConfigError(f'You cannot use NetFlow engine-id "{engine_id}" '\
-                                      f'together with NetFlow protocol version "{version}"!')
-            else:
-                regex_filter = re.compile(v9v10_filter)
-                if not regex_filter.search(flow_config['netflow']['engine_id']):
-                    raise ConfigError(f'Can not use NetFlow engine-id "{engine_id}" together '\
-                                      f'with NetFlow protocol version "{version}"!')
+    # check if vrf is defined for netflow
+    netflow_vrf = None
+    if 'vrf' in flow_config:
+        netflow_vrf = flow_config['vrf']
+
+    # Check if configured netflow server source-address exist in the system
+    # Check if configured netflow server source-address matches protocol of server
+    # Check if configured netflow server source-interface exists
+    for server, data in flow_config['netflow']['server'].items():
+        if 'source_address' in data and 'source_interface' in data:
+            raise ConfigError(
+                f'Configured "netflow server {server}" cannot have both "source-address" and "source-interface" fields'
+            )
+
+        if 'source_address' in data:
+            if not is_addr_assigned(data['source_address'], netflow_vrf):
+                raise ConfigError(
+                    f'Configured "netflow server {server} source-address {data["source_address"]}" does not exist on the system!'
+                )
+            if (
+                ip_interface(server).version
+                != ip_interface(data['source_address']).version
+            ):
+                raise ConfigError(
+                    f'Configured "netflow server {server} source-address {data["source_address"]}" protocol doesn\'t match server protocol'
+                )
+
+        if 'source_interface' in data:
+            verify_interface_exists(
+                flow_config, data['source_interface'], warning_only=True
+            )
+
+    # Check if engine-id compatible with selected protocol version
+    if 'engine_id' in flow_config['netflow']:
+        v5_filter = '^(\d|[1-9]\d|1\d{2}|2[0-4]\d|25[0-5]):(\d|[1-9]\d|1\d{2}|2[0-4]\d|25[0-5])$'
+        v9v10_filter = '^(\d|[1-9]\d{1,8}|[1-3]\d{9}|4[01]\d{8}|42[0-8]\d{7}|429[0-3]\d{6}|4294[0-8]\d{5}|42949[0-5]\d{4}|429496[0-6]\d{3}|4294967[01]\d{2}|42949672[0-8]\d|429496729[0-5])$'
+        engine_id = flow_config['netflow']['engine_id']
+        version = flow_config['netflow']['version']
+
+        if flow_config['netflow']['version'] == '5':
+            regex_filter = re.compile(v5_filter)
+            if not regex_filter.search(engine_id):
+                raise ConfigError(
+                    f'You cannot use NetFlow engine-id "{engine_id}" '
+                    f'together with NetFlow protocol version "{version}"!'
+                )
+        else:
+            regex_filter = re.compile(v9v10_filter)
+            if not regex_filter.search(flow_config['netflow']['engine_id']):
+                raise ConfigError(
+                    f'Can not use NetFlow engine-id "{engine_id}" together '
+                    f'with NetFlow protocol version "{version}"!'
+                )
 
     # return True if all checks were passed
     return True
 
+
 def generate(flow_config):
     if not flow_config:
+        if os.path.exists(ipt_netflow_conf_path):
+            os.unlink(ipt_netflow_conf_path)
         return None
 
-    render(uacctd_conf_path, 'pmacct/uacctd.conf.j2', flow_config)
-    render(systemd_override, 'pmacct/override.conf.j2', flow_config)
-    # Reload systemd manager configuration
-    call('systemctl daemon-reload')
+    prev_config = read_file(ipt_netflow_conf_path, defaultonfailure='')
 
-def apply(flow_config):
-    # Check if flow-accounting was removed and define command
-    if not flow_config:
-        _nftables_config([], 'ingress')
-        _nftables_config([], 'egress')
+    render(ipt_netflow_conf_path, 'ipt-netflow/ipt_NETFLOW.conf.j2', flow_config)
+
+    new_config = read_file(ipt_netflow_conf_path, defaultonfailure='')
 
-        # Stop flow-accounting daemon and remove configuration file
-        call(f'systemctl stop {systemd_service}')
-        if os.path.exists(uacctd_conf_path):
-            os.unlink(uacctd_conf_path)
+    global need_reload
+    need_reload = prev_config != new_config
 
-        # must be done after systemctl
-        _nftables_trigger_setup('delete')
 
+def apply(flow_config):
+    # When reloading module we need to first remove
+    # all iptables usage of ipt_NETFLOW
+    # When flow_config is disabled everything should be cleaned-up too
+    if need_reload or not flow_config:
+        ipt_netflow.stop()
+
+    if not flow_config:
+        if os.path.exists(ipt_netflow_conf_path):
+            os.unlink(ipt_netflow_conf_path)
         return
 
-    # Start/reload flow-accounting daemon
-    call(f'systemctl restart {systemd_service}')
+    ingress_interfaces = []
+    egress_interfaces = []
 
-    # configure nftables rules for defined interfaces
-    if 'interface' in flow_config:
-        _nftables_config(flow_config['interface'], 'ingress', flow_config['packet_length'])
+    # configure iptables for defined interfaces
+    if 'interface' in flow_config['netflow']:
+        ingress_interfaces = flow_config['netflow']['interface']
 
         # configure egress the same way if configured otherwise remove it
         if 'enable_egress' in flow_config:
-            _nftables_config(flow_config['interface'], 'egress', flow_config['packet_length'])
-        else:
-            _nftables_config([], 'egress')
+            egress_interfaces = ingress_interfaces
 
-    # add a trigger for signal processing
-    _nftables_trigger_setup('add')
+    if need_reload:
+        ipt_netflow.start(ingress_interfaces, egress_interfaces)
+    else:
+        ipt_netflow.set_watched_iptables_interfaces(
+            ingress_interfaces, egress_interfaces
+        )
 
 
 if __name__ == '__main__':
diff --git a/src/migration-scripts/flow-accounting/2-to-3 b/src/migration-scripts/flow-accounting/2-to-3
new file mode 100644
index 000000000..48292fc1d
--- /dev/null
+++ b/src/migration-scripts/flow-accounting/2-to-3
@@ -0,0 +1,65 @@
+# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this library.  If not, see <http://www.gnu.org/licenses/>.
+
+# migrate from pmacct to ipt-NETFLOW:
+# Remove 'timeout' subtree, 'buffer-size', 'disable-imt', 'packet-length' and
+# 'syslog-facility' from 'system flow-accounting'
+#
+# Move "system flow-accounting interface" to "system flow-accounting netflow interface"
+#
+# Remove "system flow-accounting source" and add it to each server
+#   under "system flow-accounting netflow server SERVER source-address"
+
+
+from vyos.configtree import ConfigTree
+
+base = ['system', 'flow-accounting']
+remove_keys = [
+    ['buffer-size'],
+    ['disable-imt'],
+    ['netflow', 'timeout'],
+    ['packet-length'],
+    ['syslog-facility'],
+]
+
+def migrate(config: ConfigTree) -> None:
+    if not config.exists(base):
+        # Nothing to do
+        return
+
+    # Remove not needed pmacct fields
+    for k in remove_keys:
+        p = base + k
+        if config.exists(p):
+            config.delete(p)
+
+    # Move "system flow-accounting interface" -> "system flow-accounting netflow interface"
+    if config.exists(base + ['interface']):
+        config.copy(base + ['interface'], base + ['netflow', 'interface'])
+        config.delete(base + ['interface'])
+
+    # Remove old "source-address", add it to each server as "source-address"
+    source_prev_path = base + ['netflow', 'source-address']
+    if config.exists(source_prev_path):
+        source_value = config.return_value(source_prev_path)
+        config.delete(source_prev_path)
+
+        # So we loose 'source-address' value if there are no servers
+        # configured, but it is not valid configuration if there
+        # is no server, so it shouldn't be a problem
+        if config.exists(base + ['netflow', 'server']):
+            path = base + ['netflow', 'server']
+            for server in config.list_nodes(path):
+                config.set(path + [server, 'source-address'], source_value)
diff --git a/src/op_mode/flow_accounting_op.py b/src/op_mode/flow_accounting_op.py
index f8aabc1ee..0c3184fc1 100755
--- a/src/op_mode/flow_accounting_op.py
+++ b/src/op_mode/flow_accounting_op.py
@@ -18,18 +18,16 @@ import sys
 import argparse
 import re
 import ipaddress
-import os.path
 
 from tabulate import tabulate
-from json import loads
-from vyos.utils.commit import commit_in_progress
+from vyos.utils.kernel import is_module_loaded
 from vyos.utils.process import cmd
-from vyos.utils.process import run
 from vyos.logger import syslog
+from vyos.configquery import ConfigTreeQuery
+from vyos import ipt_netflow
 
 # some default values
-uacctd_pidfile = '/var/run/uacctd.pid'
-uacctd_pipefile = '/tmp/uacctd.pipe'
+flows_dump_path = '/proc/net/stat/ipt_netflow_flows'
 
 def parse_port(port):
     try:
@@ -45,7 +43,7 @@ def parse_ports(arg):
     if re.match(r'^\d+$', arg):
         # Single port
         port = parse_port(arg)
-        return {"type": "single", "value": port}
+        return {"type": "single", "values": (port,)}
     elif re.match(r'^\d+\-\d+$', arg):
         # Port range
         ports = arg.split("-")
@@ -53,12 +51,12 @@ def parse_ports(arg):
         if ports[0] > ports[1]:
             raise ValueError("Malformed port range \'{0}\': lower end is greater than the higher".format(arg))
         else:
-            return {"type": "range", "value": (ports[0], ports[1])}
+            return {"type": "range", "values": range(ports[0], ports[1] + 1)}
     elif re.match(r'^\d+,.*\d$', arg):
         # Port list
-        ports = re.split(r',+', arg) # This allows duplicate commad like '1,,2,3,4'
+        ports = re.split(r',+', arg)  # This allows duplicate commas like '1,,2,3,4'
         ports = list(map(parse_port, ports))
-        return {"type": "list", "value": ports}
+        return {"type": "list", "values": ports}
     else:
         raise ValueError("Malformed port spec \'{0}\'".format(arg))
 
@@ -69,9 +67,8 @@ def check_host(host):
         raise ValueError("Invalid host \'{}\', must be a valid IP or IPv6 address".format(host))
 
 # check if flow-accounting running
-def _uacctd_running():
-    command = 'systemctl status uacctd.service > /dev/null'
-    return run(command) == 0
+def _netflow_running():
+    return is_module_loaded(ipt_netflow.module_name)
 
 
 # get list of interfaces
@@ -95,20 +92,56 @@ def _get_ifaces_dict():
 
 # get list of flows
 def _get_flows_list():
-    # run command to get flows list
-    out = cmd(f'/usr/bin/pmacct -s -O json -T flows -p {uacctd_pipefile}',
-              message='Failed to get flows list')
+    # File format:
+    # When MAC disabled:
+    # # hash a dev:i,o proto src:ip,port dst:ip,port nexthop tos,tcpflags,options,tcpoptions packets bytes ts:first,last
+    # 1 c06c 0 4,-1 1 10.2.0.7,0 10.1.0.5,0 0.0.0.0 0,0,0,0 186 15624 92261,131
+    # 2 1e3ca 0 3,-1 1 10.1.0.5,0 10.2.0.7,2048 0.0.0.0 0,0,0,0 186 15624 92261,132
+
+    # When MAC enabled + VLAN fix:
+    # hash a dev:i,o mac:src,dst vlan type proto src:ip,port dst:ip,port nexthop tos,tcpflags,options,tcpoptions packets bytes ts:first,last
+    # 1 11a41 0 4,-1 0c:27:1f:55:00:00,0c:e8:b1:71:00:02 - 0800 1 10.2.0.7,0 10.1.0.5,0 0.0.0.0 0,0,0,0 1182 99288 591502,529
+    # 2 13bc5 0 4,-1 0c:27:1f:55:00:00,0c:e8:b1:71:00:02 - 0800 1 10.2.0.7,0 10.2.0.1,2048 0.0.0.0 0,0,0,0 577 48468 590831,1006
+    # 3 166dd 0 3,-1 0c:f1:0a:d5:00:00,0c:e8:b1:71:00:01 - 0800 1 10.1.0.5,0 10.2.0.7,2048 0.0.0.0 0,0,0,0 1182 99288 591502,529
 
-    # read output
-    flows_out = out.splitlines()
 
-    # make a list with flows
     flows_list = []
-    for flow_line in flows_out:
-        try:
-            flows_list.append(loads(flow_line))
-        except Exception as err:
-            syslog.error('Unable to read flow info: {}'.format(err))
+    with open(flows_dump_path) as f:
+        headers = f.readline()
+        headers = headers.split()
+        for i, h in enumerate(headers):
+
+            if ',' in h and ':' not in h:
+                h = 'extra:' + h
+
+            if ':' in h:
+                key, subkeys = h.split(':', 1)
+                headers[i] = {'key': key, 'subkeys': subkeys.split(',')}
+
+        linenum = 1
+        for flow_line in f:
+            linenum += 1
+            flow_dict = {}
+            flow_line = flow_line.split()
+            if len(flow_line) != len(headers):
+                syslog.error(
+                    f'Unexpected number of elements in {flows_dump_path}, line {linenum}'
+                )
+                continue
+            for i, val in enumerate(flow_line):
+                if isinstance(headers[i], str):
+                    flow_dict[headers[i]] = val
+                elif isinstance(headers[i], dict):
+                    val = val.split(',')
+                    if len(val) != len(headers[i]['subkeys']):
+                        syslog.error(
+                            f"Unexpected number of elements in {flows_dump_path} in column {headers[i]['key']} in line {linenum}"
+                        )
+                        continue
+                    flow_dict[headers[i]['key']] = dict(zip(headers[i]['subkeys'], val))
+                else:
+                    assert False, "Unexpected type of header"
+            flows_list.append(flow_dict)
 
     # return list of flows
     return flows_list
@@ -119,12 +152,15 @@ def _flows_filter(flows, ifaces):
     # predefine filtered flows list
     flows_filtered = []
 
+    def _iface_to_str(iface):
+        if int(iface) in ifaces:
+            return ifaces[int(iface)]
+        return 'unknown'
+
     # add interface names to flows
     for flow in flows:
-        if flow['iface_in'] in ifaces:
-            flow['iface_in_name'] = ifaces[flow['iface_in']]
-        else:
-            flow['iface_in_name'] = 'unknown'
+        flow['iface_in_name'] = _iface_to_str(flow['dev']['i'])
+        flow['iface_out_name'] = _iface_to_str(flow['dev']['o'])
 
     # iterate through flows list
     for flow in flows:
@@ -134,16 +170,19 @@ def _flows_filter(flows, ifaces):
                 continue
         # filter by host
         if cmd_args.host:
-            if flow['ip_src'] != cmd_args.host and flow['ip_dst'] != cmd_args.host:
+            if (
+                flow['src']['ip'] != cmd_args.host
+                and flow['dst']['ip'] != cmd_args.host
+            ):
                 continue
         # filter by ports
         if cmd_args.ports:
-            if cmd_args.ports['type'] == 'single':
-                if flow['port_src'] != cmd_args.ports['value'] and flow['port_dst'] != cmd_args.ports['value']:
-                    continue
-            else:
-                if flow['port_src'] not in cmd_args.ports['value'] and flow['port_dst'] not in cmd_args.ports['value']:
-                    continue
+            # for 'single' it is a tuple with one value, for 'list' - list of ports, for range - range of ports
+            if (
+                int(flow['src']['port']) not in cmd_args.ports['values']
+                and int(flow['dst']['port']) not in cmd_args.ports['values']
+            ):
+                continue
         # add filtered flows to new list
         flows_filtered.append(flow)
 
@@ -159,23 +198,36 @@ def _flows_filter(flows, ifaces):
 # print flow table
 def _flows_table_print(flows):
     # define headers and body
-    table_headers = ['IN_IFACE', 'SRC_MAC', 'DST_MAC', 'SRC_IP', 'DST_IP', 'SRC_PORT', 'DST_PORT', 'PROTOCOL', 'TOS', 'PACKETS', 'FLOWS', 'BYTES']
+    table_headers = [
+        'IN_IFACE',
+        'SRC_MAC',
+        'DST_MAC',
+        'SRC_IP',
+        'DST_IP',
+        'SRC_PORT',
+        'DST_PORT',
+        'PROTOCOL',
+        'TOS',
+        'PACKETS',
+        # 'FLOWS', # What was here in pmacct?
+        'BYTES',
+    ]
     table_body = []
     # convert flows to list
     for flow in flows:
         table_line = [
             flow.get('iface_in_name'),
-            flow.get('mac_src'),
-            flow.get('mac_dst'),
-            flow.get('ip_src'),
-            flow.get('ip_dst'),
-            flow.get('port_src'),
-            flow.get('port_dst'),
-            flow.get('ip_proto'),
-            flow.get('tos'),
+            flow.get('mac', {}).get('src'),
+            flow.get('mac', {}).get('dst'),
+            flow.get('src', {}).get('ip'),
+            flow.get('dst', {}).get('ip'),
+            flow.get('src', {}).get('port'),
+            flow.get('dst', {}).get('port'),
+            flow.get('proto'),
+            flow.get('extra', {}).get('tos'),
             flow.get('packets'),
-            flow.get('flows'),
-            flow.get('bytes')
+            # flow.get('flows'),
+            flow.get('bytes'),
         ]
         table_body.append(table_line)
     # configure and fill table
@@ -190,21 +242,37 @@ def _flows_table_print(flows):
         sys.exit(0)
 
 
-# check if in-memory table is active
-def _check_imt():
-    if not os.path.exists(uacctd_pipefile):
-        print("In-memory table is not available")
-        sys.exit(1)
-
-
 # define program arguments
 cmd_args_parser = argparse.ArgumentParser(description='show flow-accounting')
-cmd_args_parser.add_argument('--action', choices=['show', 'clear', 'restart'], required=True, help='command to flow-accounting daemon')
-cmd_args_parser.add_argument('--filter', choices=['interface', 'host', 'ports', 'top'], required=False,  nargs='*', help='filter flows to display')
-cmd_args_parser.add_argument('--interface', required=False, help='interface name for output filtration')
-cmd_args_parser.add_argument('--host', type=str, required=False, help='host address for output filtering')
-cmd_args_parser.add_argument('--ports', type=str, required=False, help='port number, range or list for output filtering')
-cmd_args_parser.add_argument('--top', type=int, required=False, help='top records for output filtering')
+# 'clear' and 'restart' are not implemented
+cmd_args_parser.add_argument(
+    '--action',
+    choices=['show', 'restart'],
+    default='show',
+    help='show stat or restart module',
+)
+cmd_args_parser.add_argument(
+    '--filter',
+    choices=['interface', 'host', 'ports', 'top'],
+    required=False,
+    nargs='*',
+    help='filter flows to display',
+)
+cmd_args_parser.add_argument(
+    '--interface', required=False, help='interface name for output filtration'
+)
+cmd_args_parser.add_argument(
+    '--host', type=str, required=False, help='host address for output filtering'
+)
+cmd_args_parser.add_argument(
+    '--ports',
+    type=str,
+    required=False,
+    help='port number, range or list for output filtering',
+)
+cmd_args_parser.add_argument(
+    '--top', type=int, required=False, help='top records for output filtering'
+)
 # parse arguments
 cmd_args = cmd_args_parser.parse_args()
 
@@ -219,30 +287,13 @@ except ValueError as e:
     sys.exit(1)
 
 # main logic
-# do nothing if uacctd daemon is not running
-if not _uacctd_running():
+# do nothing if ipt_NETFLOW is not active
+if not _netflow_running():
     print("flow-accounting is not active")
     sys.exit(1)
 
-# restart pmacct daemon
-if cmd_args.action == 'restart':
-    if commit_in_progress():
-        print('Cannot restart flow-accounting while a commit is in progress')
-        exit(1)
-    # run command to restart flow-accounting
-    cmd('systemctl restart uacctd.service',
-        message='Failed to restart flow-accounting')
-
-# clear in-memory collected flows
-if cmd_args.action == 'clear':
-    _check_imt()
-    # run command to clear flows
-    cmd(f'/usr/bin/pmacct -e -p {uacctd_pipefile}',
-        message='Failed to clear flows')
-
 # show table with flows
 if cmd_args.action == 'show':
-    _check_imt()
     # get interfaces index and names
     ifaces_dict = _get_ifaces_dict()
     # get flows
@@ -254,4 +305,22 @@ if cmd_args.action == 'show':
     # print flows
     _flows_table_print(tabledata)
 
+if cmd_args.action == 'restart':
+    ipt_netflow.stop()
+
+    # get needed interfaces
+    conf = ConfigTreeQuery()
+    config_path = ['system', 'flow-accounting']
+    if not conf.exists(config_path + ['netflow', 'interface']):
+        print("Flow accounting not configured, exiting")
+        sys.exit(1)
+
+    ingress_interfaces = conf.values(config_path + ['netflow', 'interface'])
+    if conf.exists(config_path + ['enable-egress']):
+        egress_interfaces = ingress_interfaces
+    else:
+        egress_interfaces = []
+
+    ipt_netflow.start(ingress_interfaces, egress_interfaces)
+
 sys.exit(0)