diff --git a/meta/runtime.yml b/meta/runtime.yml
index 59c24bdfaa..51b6538c78 100644
--- a/meta/runtime.yml
+++ b/meta/runtime.yml
@@ -297,5 +297,9 @@ action_groups:
     - azure.azcollection.azure_rm_webappslot
     - azure.azcollection.azure_rm_webappvnetconnection
     - azure.azcollection.azure_rm_webappvnetconnection_info
+    - azure.azcollection.azure_rm_networkwatcher
+    - azure.azcollection.azure_rm_networkwatcher_info
+    - azure.azcollection.azure_rm_networkflowlog
+    - azure.azcollection.azure_rm_networkflowlog_info
     - azure.azcollection.azure_rm_capacityreservationgroup
     - azure.azcollection.azure_rm_capacityreservationgroup_info
diff --git a/plugins/modules/azure_rm_networkflowlog.py b/plugins/modules/azure_rm_networkflowlog.py
new file mode 100644
index 0000000000..4a0cabf06c
--- /dev/null
+++ b/plugins/modules/azure_rm_networkflowlog.py
@@ -0,0 +1,477 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2024 xuzhang3 (@xuzhang3), Fred-sun (@Fred-sun)
+#
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = '''
+---
+module: azure_rm_networkflowlog
+version_added: "2.5.0"
+short_description: Manage the network flow logs
+description:
+    - Create, update or delete the network flow logs.
+options:
+    resource_group:
+        description:
+            - Name of resource group.
+        required: true
+        type: str
+    location:
+        description:
+            - Valid Azure location. Defaults to location of the resource group.
+        type: str
+    name:
+        description:
+            - The name of the network flow logs.
+        required: true
+        type: str
+    network_watcher_name:
+        description:
+            - The name of the network watcher.
+        type: str
+        required: true
+    target_resource_id:
+        description:
+            -  ID of network security group to which flow log will be applied.
+        type: str
+    storage_id:
+        description:
+            - ID of the storage account which is used to store the flow log.
+        type: str
+    enabled:
+        description:
+            - Flag to enable/disable flow logging.
+        type: bool
+    retention_policy:
+        description:
+            - Parameters that define the retention policy for flow log.
+        type: dict
+        suboptions:
+            days:
+                description:
+                    - Number of days to retain flow log records.
+                type: int
+            enabled:
+                description:
+                    - Flag to enable/disable retention.
+                type: bool
+    flow_analytics_configuration:
+        description:
+            - Parameters that define the configuration of traffic analytics.
+        type: dict
+        suboptions:
+            network_watcher_flow_analytics_configuration:
+                description:
+                    - Parameters that define the configuration of traffic analytics.
+                type: dict
+                suboptions:
+                    enabled:
+                        description:
+                            - Flag to enable/disable traffic analytics.
+                        type: bool
+                    workspace_id:
+                        description:
+                            - The resource guid of the attached workspace.
+                        type: str
+                    workspace_region:
+                        description:
+                            - The location of the attached workspace.
+                        type: str
+                    workspace_resource_id:
+                        description:
+                            - Resource Id of the attached workspace.
+                        type: str
+                    traffic_analytics_interval:
+                        description:
+                            - The interval in minutes which would decide how frequently TA service should do flow analytics.
+                        type: int
+                        choices:
+                            - 10
+                            - 60
+    state:
+        description:
+            - State of the Flow Logs. Use C(present) to create or update and C(absent) to delete.
+        default: present
+        type: str
+        choices:
+            - absent
+            - present
+
+extends_documentation_fragment:
+    - azure.azcollection.azure
+    - azure.azcollection.azure_tags
+
+author:
+    - xuzhang3 (@xuzhang3)
+    - Fred-sun (@Fred-sun)
+
+'''
+
+EXAMPLES = '''
+- name: Create network flow log
+  azure_rm_networkflowlog:
+    resource_group: NetworkWatcherRG
+    network_watcher_name: NetworkWatcher_eastus
+    name: xz3mlwvnet-xz3mlwaiserv-flowlog02
+    enabled: false
+    flow_analytics_configuration:
+      network_watcher_flow_analytics_configuration:
+        enabled: false
+        traffic_analytics_interval: 60
+        workspace_id: 7c16a8dd-b983-4f75-b78b-a804c169306c
+        workspace_region: eastus
+        workspace_resource_id: "/subscriptions/xxx-xxx/resourceGroups/DefaultRG-EUS/providers/Microsoft.OperationalInsights/workspaces/DeWorkspace-0-EUS"
+    retention_policy:
+      days: 2
+      enabled: true
+    storage_id: "/subscriptions/xxx-xxx/resourceGroups/AutoTagFunctionAppRG/providers/Microsoft.Storage/storageAccounts/autotagfunctionappr9a08"
+    target_resource_id: "/subscriptions/xxx-xxx/resourceGroups/xz3mlwaiserv/providers/Microsoft.Network/virtualNetworks/xz3mlwvnet"
+    tags:
+      key2: value2
+      key5: value5
+
+- name: Delete a Flow Logs
+  azure_rm_networkflowlog:
+    resource_group: myResourceGroup
+    network_watcher_name: testwatcher
+    name: myNetflowlog
+    state: absent
+'''
+RETURN = '''
+state:
+    description:
+        - The facts of the network flow logs.
+    returned: always
+    type: complex
+    contains:
+        resource_group:
+            description:
+                - The resource group.
+            type: str
+            returned: always
+            sample: NetworkWatcherRG
+        id:
+            description:
+                - Resource ID.
+            returned: always
+            type: str
+            sample: /subscriptions/xxx-xxx/resourceGroups/NetWatcherRG/providers/Microsoft.Network/networkWatchers/NetWatcher_eastus/flowLogs/xz-flowlog"
+        location:
+            description:
+                - Resource location.
+            returned: always
+            type: str
+            sample: eastus
+        name:
+            description:
+                - Resource name.
+            returned: always
+            type: str
+            sample: xz-flowlog
+        network_watcher_name:
+            description:
+                - The name of the network watcher.
+            type: str
+            returned: always
+            sample: NetWatcher_eastus
+        target_resource_id:
+            description:
+                - ID of network security group to which flow log will be applied.
+            type: str
+            returned: always
+            sample: /subscriptions/xxx-xxx/resourceGroups/xz3mlwaiserv/providers/Microsoft.Network/virtualNetworks/xz3mlwvnet"
+        storage_id:
+            description:
+                - ID of the storage account which is used to store the flow log.
+            type: str
+            returned: always
+            sample: "/subscriptions/xxx-xxx/resourceGroups/AutoTagFunctionAppRG/providers/Microsoft.Storage/storageAccounts/autotagfunc01"
+        enanbled:
+            description:
+                - Flag to enable/disable flow logging.
+            type: str
+            returned: always
+            sample: true
+        retention_policy:
+            description:
+                - Parameters that define the retention policy for flow log.
+            type: complex
+            returned: always
+            contains:
+                day:
+                    description:
+                        - Number of days to retain flow log records.
+                    type: int
+                    returned: always
+                    sample: 0
+                enabled:
+                    description:
+                        - Flag to enable/disable retention.
+                    type: bool
+                    returned: always
+                    sample: false
+        flow_analytics_configuration:
+            description:
+                - Parameters that define the configuration of traffic analytics.
+            type: complex
+            returned: always
+            contains:
+                network_watcher_flow_analytics_configuration:
+                    description:
+                        - Parameters that define the configuration of traffic analytics.
+                    type: complex
+                    returned: always
+                    contains:
+                        enabled:
+                            description:
+                                - Flag to enable/disable traffic analytics.
+                            type: bool
+                            returned: always
+                            sample: true
+                        workspace_id:
+                            description:
+                                - The resource guid of the attached workspace.
+                            type: str
+                            returned: always
+                            sample: 7c16a8dd-b983-4f75-b78b-a804c169306c
+                        workspace_region:
+                            description:
+                                - The location of the attached workspace.
+                            type: str
+                            returned: always
+                            sample: eastus
+                        workspace_resource_id:
+                            description:
+                                - Resource Id of the attached workspace.
+                            type: str
+                            returned: always
+                            sample: /subscriptions/xxx-xxx/resourceGroups/DefaulUS/providers/Microsoft.OperationalInsights/workspaces/DefaultWorkspace-0-EUS"
+                        traffic_analytics_interval:
+                            description:
+                                - The interval in minutes which would decide how frequently TA service should do flow analytics.
+                            type: str
+                            returned: always
+                            sample: 60
+        tags:
+            description:
+                - Resource tags.
+            returned: always
+            type: dict
+            sample: { 'key1':'value1' }
+        type:
+            description:
+                - Resource type.
+            returned: always
+            type: str
+            sample: "Microsoft.Network/networkWatchers/flowLogs"
+        provisioning_state:
+            description:
+                - The provisioning state of the network flow logs resource.
+            type: str
+            returned: always
+            sample: Succeeded
+'''
+
+try:
+    from azure.core.exceptions import ResourceNotFoundError
+except ImportError:
+    # This is handled in azure_rm_common
+    pass
+
+from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common_ext import AzureRMModuleBaseExt
+
+
+class AzureRMNetworkFlowLog(AzureRMModuleBaseExt):
+
+    def __init__(self):
+
+        self.module_arg_spec = dict(
+            resource_group=dict(type='str', required=True),
+            network_watcher_name=dict(type='str', required=True),
+            name=dict(type='str', required=True),
+            state=dict(type='str', default='present', choices=['present', 'absent']),
+            location=dict(type='str'),
+            target_resource_id=dict(type='str'),
+            storage_id=dict(type='str'),
+            enabled=dict(type='bool'),
+            retention_policy=dict(
+                type='dict',
+                options=dict(
+                    days=dict(type='int'),
+                    enabled=dict(type='bool'),
+                ),
+            ),
+            flow_analytics_configuration=dict(
+                type='dict',
+                options=dict(
+                    network_watcher_flow_analytics_configuration=dict(
+                        type='dict',
+                        options=dict(
+                            enabled=dict(type='bool'),
+                            workspace_id=dict(type='str'),
+                            workspace_region=dict(type='str'),
+                            workspace_resource_id=dict(type='str'),
+                            traffic_analytics_interval=dict(type='int', choices=[10, 60])
+                        )
+                    )
+                )
+            ),
+        )
+
+        self.resource_group = None
+        self.network_watcher_name = None
+        self.name = None
+        self.state = None
+        self.location = None
+        self.tags = None
+        self.body = dict()
+
+        self.results = dict(
+            changed=False,
+            state=dict()
+        )
+
+        super(AzureRMNetworkFlowLog, self).__init__(self.module_arg_spec,
+                                                    supports_tags=True,
+                                                    supports_check_mode=True)
+
+    def exec_module(self, **kwargs):
+
+        for key in list(self.module_arg_spec.keys()) + ['tags']:
+            if hasattr(self, key):
+                setattr(self, key, kwargs[key])
+            elif kwargs[key] is not None:
+                self.body[key] = kwargs[key]
+
+        resource_group = self.get_resource_group(self.resource_group)
+        if self.body.get('location') is None:
+            # Set default location
+            self.body['location'] = resource_group.location
+        self.body['tags'] = self.tags
+
+        changed = False
+        results = dict()
+
+        old_response = self.get_by_name()
+
+        if old_response is not None:
+            if self.state == 'present':
+                if self.body.get('retention_policy') is not None and\
+                   not self.default_compare({}, self.body.get('retention_policy'), old_response.get('retention_policy'), '', dict(compare=[])):
+                    changed = True
+                elif (self.body.get('flow_analytics_configuration') is not None and not self.default_compare(
+                        {}, self.body['flow_analytics_configuration'], old_response['flow_analytics_configuration'], '', dict(compare=[]))):
+                    changed = True
+
+                elif self.body.get('enabled') is not None and bool(self.body['enabled']) != bool(old_response.get('enabled')):
+                    changed = True
+                if changed:
+                    results = self.create_or_update(self.body)
+                else:
+                    results = old_response
+
+                update_tags, new_tags = self.update_tags(old_response['tags'])
+                if update_tags:
+                    changed = True
+                    if not self.check_mode:
+                        results = self.update_flowlog_tags(new_tags)
+            else:
+                changed = True
+                if not self.check_mode:
+                    results = self.delete_flowlog()
+        else:
+            if self.state == 'present':
+                changed = True
+                if not self.check_mode:
+                    results = self.create_or_update(self.body)
+            else:
+                changed = False
+                self.log("The Flow Log is not exists")
+
+        self.results['changed'] = changed
+        self.results['state'] = results
+
+        return self.results
+
+    def get_by_name(self):
+        response = None
+        try:
+            response = self.network_client.flow_logs.get(self.resource_group, self.network_watcher_name, self.name)
+
+        except ResourceNotFoundError as exec:
+            self.log("Failed to get network flow log, Exception as {0}".format(exec))
+
+        return self.to_dict(response)
+
+    def create_or_update(self, body):
+        response = None
+        try:
+            response = self.network_client.flow_logs.begin_create_or_update(self.resource_group, self.network_watcher_name, self.name, body)
+            response = self.get_poller_result(response)
+        except Exception as exc:
+            self.fail("Error creating Flow Log {0} - {1}".format(self.name, str(exc)))
+
+        return self.to_dict(response)
+
+    def update_flowlog_tags(self, tags):
+        response = None
+        try:
+            response = self.network_client.flow_logs.update_tags(self.resource_group, self.network_watcher_name, self.name, dict(tags=tags))
+        except Exception as exc:
+            self.fail("Error updating Flow Logs {0} - {1}".format(self.name, str(exc)))
+        return self.to_dict(response)
+
+    def delete_flowlog(self):
+        try:
+            self.network_client.flow_logs.begin_delete(self.resource_group, self.network_watcher_name, self.name)
+        except Exception as exc:
+            self.fail("Error deleting Flow Logs {0} - {1}".format(self.name, str(exc)))
+
+    def to_dict(self, body):
+        results = dict()
+        if body is not None:
+            results = dict(
+                resource_group=self.resource_group,
+                network_watcher_name=self.network_watcher_name,
+                id=body.id,
+                name=body.name,
+                location=body.location,
+                tags=body.tags,
+                type=body.type,
+                provisioning_state=body.provisioning_state,
+                target_resource_id=body.target_resource_id,
+                storage_id=body.storage_id,
+                enabled=body.enabled,
+                retention_policy=dict(),
+                flow_analytics_configuration=dict()
+            )
+            if body.retention_policy is not None:
+                results['retention_policy']['days'] = body.retention_policy.days
+                results['retention_policy']['enabled'] = body.retention_policy.enabled
+            if body.flow_analytics_configuration is not None:
+                results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration'] = dict()
+                if body.flow_analytics_configuration.network_watcher_flow_analytics_configuration is not None:
+                    new_config = body.flow_analytics_configuration.network_watcher_flow_analytics_configuration
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['enabled'] = new_config.enabled
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['workspace_id'] = new_config.workspace_id
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['workspace_region'] = new_config.workspace_region
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['workspace_resource_id'] = \
+                        new_config.workspace_resource_id
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['traffic_analytics_interval'] = \
+                        new_config.traffic_analytics_interval
+
+            return results
+        return None
+
+
+def main():
+    AzureRMNetworkFlowLog()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/plugins/modules/azure_rm_networkflowlog_info.py b/plugins/modules/azure_rm_networkflowlog_info.py
new file mode 100644
index 0000000000..860a7cfe32
--- /dev/null
+++ b/plugins/modules/azure_rm_networkflowlog_info.py
@@ -0,0 +1,309 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2024 xuzhang3 (@xuzhang3), Fred-sun (@Fred-sun)
+#
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = '''
+---
+module: azure_rm_networkflowlog_info
+
+version_added: "2.5.0"
+
+short_description: Get or list the network flow logs
+
+description:
+    - Get or list the network flow logs facts.
+
+options:
+    resource_group:
+        description:
+            - The name of the resource group.
+        type: str
+        required: true
+    network_watcher_name:
+        description:
+            - Name of the network watcher.
+        type: str
+        required: true
+    name:
+        description:
+            - Name of the network flow logs.
+        type: str
+    tags:
+        description:
+            - Limit results by providing a list of tags. Format tags as 'key' or 'key:value'.
+        type: list
+        elements: str
+
+extends_documentation_fragment:
+    - azure.azcollection.azure
+
+author:
+    - xuzhang3 (@xuzhang3)
+    - Fred-sun (@Fred-sun)
+
+'''
+
+EXAMPLES = '''
+- name: Get the network watcher facts
+  azure_rm_networkflowlog_info:
+    resource_group: myResourceGroup
+    network_watcher_name: mywatcher01
+    name: flowlogname
+
+- name: list the network flow logs and filter by tags
+  azure_rm_networkflowlog_info:
+    resource_group: myResourceGroup
+    network_watcher_name: mywatcher01
+    tags:
+      - key1
+'''
+
+RETURN = '''
+flow_logs:
+    description:
+        - The facts of the network flow logs.
+    returned: always
+    type: complex
+    contains:
+        resource_group:
+            description:
+                - The resource group.
+            type: str
+            returned: always
+            sample: NetworkWatcherRG
+        id:
+            description:
+                - Resource ID.
+            returned: always
+            type: str
+            sample: /subscriptions/xxx-xxx/resourceGroups/NetWatcherRG/providers/Microsoft.Network/networkWatchers/NetWatcher_eastus/flowLogs/xz-flowlog"
+        location:
+            description:
+                - Resource location.
+            returned: always
+            type: str
+            sample: eastus
+        name:
+            description:
+                - Resource name.
+            returned: always
+            type: str
+            sample: xz-flowlog
+        network_watcher_name:
+            description:
+                - The name of the network watcher.
+            type: str
+            returned: always
+            sample: NetWatcher_eastus
+        target_resource_id:
+            description:
+                - ID of network security group to which flow log will be applied.
+            type: str
+            returned: always
+            sample: /subscriptions/xxx-xxx/resourceGroups/xz3mlwaiserv/providers/Microsoft.Network/virtualNetworks/xz3mlwvnet"
+        storage_id:
+            description:
+                - ID of the storage account which is used to store the flow log.
+            type: str
+            returned: always
+            sample: "/subscriptions/xxx-xxx/resourceGroups/AutoTagFunctionAppRG/providers/Microsoft.Storage/storageAccounts/autotagfunc01"
+        enanbled:
+            description:
+                - Flag to enable/disable flow logging.
+            type: str
+            returned: always
+            sample: true
+        retention_policy:
+            description:
+                - Parameters that define the retention policy for flow log.
+            type: complex
+            returned: always
+            contains:
+                day:
+                    description:
+                        - Number of days to retain flow log records.
+                    type: int
+                    returned: always
+                    sample: 0
+                enabled:
+                    description:
+                        - Flag to enable/disable retention.
+                    type: bool
+                    returned: always
+                    sample: false
+        flow_analytics_configuration:
+            description:
+                - Parameters that define the configuration of traffic analytics.
+            type: complex
+            returned: always
+            contains:
+                network_watcher_flow_analytics_configuration:
+                    description:
+                        - Parameters that define the configuration of traffic analytics.
+                    type: complex
+                    returned: always
+                    contains:
+                        enabled:
+                            description:
+                                - Flag to enable/disable traffic analytics.
+                            type: bool
+                            returned: always
+                            sample: true
+                        workspace_id:
+                            description:
+                                - The resource guid of the attached workspace.
+                            type: str
+                            returned: always
+                            sample: 7c16a8dd-b983-4f75-b78b-a804c169306c
+                        workspace_region:
+                            description:
+                                - The location of the attached workspace.
+                            type: str
+                            returned: always
+                            sample: eastus
+                        workspace_resource_id:
+                            description:
+                                - Resource Id of the attached workspace.
+                            type: str
+                            returned: always
+                            sample: /subscriptions/xxx-xxx/resourceGroups/DefaulUS/providers/Microsoft.OperationalInsights/workspaces/DefaultWorkspace-0-EUS"
+                        traffic_analytics_interval:
+                            description:
+                                - The interval in minutes which would decide how frequently TA service should do flow analytics.
+                            type: str
+                            returned: always
+                            sample: 60
+        tags:
+            description:
+                - Resource tags.
+            returned: always
+            type: dict
+            sample: { 'key1':'value1' }
+        type:
+            description:
+                - Resource type.
+            returned: always
+            type: str
+            sample: "Microsoft.Network/networkWatchers/flowLogs"
+        provisioning_state:
+            description:
+                - The provisioning state of the network flow logs resource.
+            type: str
+            returned: always
+            sample: Succeeded
+'''
+
+try:
+    from azure.core.exceptions import ResourceNotFoundError
+except Exception:
+    # This is handled in azure_rm_common
+    pass
+
+from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common import AzureRMModuleBase
+
+
+class AzureRMNetworkFlowLogInfo(AzureRMModuleBase):
+
+    def __init__(self):
+
+        self.module_arg_spec = dict(
+            resource_group=dict(type='str', required=True),
+            network_watcher_name=dict(type='str', required=True),
+            name=dict(type='str'),
+            tags=dict(type='list', elements='str')
+        )
+
+        self.results = dict(
+            changed=False,
+            flow_logs=[]
+        )
+
+        self.resource_group = None
+        self.network_watcher_name = None
+        self.name = None
+        self.tags = None
+
+        super(AzureRMNetworkFlowLogInfo, self).__init__(self.module_arg_spec,
+                                                        supports_check_mode=True,
+                                                        supports_tags=False,
+                                                        facts_module=True)
+
+    def exec_module(self, **kwargs):
+        for key in self.module_arg_spec:
+            setattr(self, key, kwargs[key])
+
+        if self.name:
+            response = [self.get_by_name()]
+        else:
+            response = self.list_by_network_watcher()
+
+        self.results['flow_logs'] = [self.to_dict(item) for item in response if response is not None]
+
+        return self.results
+
+    def get_by_name(self):
+        response = None
+        try:
+            response = self.network_client.flow_logs.get(self.resource_group, self.network_watcher_name, self.name)
+
+        except ResourceNotFoundError as exec:
+            self.log("Failed to get network flow logs, Exception as {0}".format(exec))
+
+        return response
+
+    def list_by_network_watcher(self):
+        response = []
+        try:
+            response = self.network_client.flow_logs.list(self.resource_group, self.network_watcher_name)
+        except Exception as exec:
+            self.log("Faild to list network flow logs by network watcher, exception as {0}".format(exec))
+        return response
+
+    def to_dict(self, body):
+        results = dict()
+        if body is not None and self.has_tags(body.tags, self.tags):
+            results = dict(
+                resource_group=self.resource_group,
+                network_watcher_name=self.network_watcher_name,
+                id=body.id,
+                name=body.name,
+                location=body.location,
+                tags=body.tags,
+                type=body.type,
+                provisioning_state=body.provisioning_state,
+                target_resource_id=body.target_resource_id,
+                storage_id=body.storage_id,
+                enabled=body.enabled,
+                retention_policy=dict(),
+                flow_analytics_configuration=dict()
+            )
+            if body.retention_policy is not None:
+                results['retention_policy']['days'] = body.retention_policy.days
+                results['retention_policy']['enabled'] = body.retention_policy.enabled
+            if body.flow_analytics_configuration is not None:
+                results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration'] = dict()
+                if body.flow_analytics_configuration.network_watcher_flow_analytics_configuration is not None:
+                    new_config = body.flow_analytics_configuration.network_watcher_flow_analytics_configuration
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['enabled'] = new_config.enabled
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['workspace_id'] = new_config.workspace_id
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['workspace_region'] = new_config.workspace_region
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['workspace_resource_id'] = \
+                        new_config.workspace_resource_id
+                    results['flow_analytics_configuration']['network_watcher_flow_analytics_configuration']['traffic_analytics_interval'] = \
+                        new_config.traffic_analytics_interval
+
+        return results
+
+
+def main():
+    AzureRMNetworkFlowLogInfo()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/plugins/modules/azure_rm_networkwatcher.py b/plugins/modules/azure_rm_networkwatcher.py
new file mode 100644
index 0000000000..3a9bed0b05
--- /dev/null
+++ b/plugins/modules/azure_rm_networkwatcher.py
@@ -0,0 +1,249 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2024 xuzhang3 (@xuzhang3), Fred-sun (@Fred-sun)
+#
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = '''
+---
+module: azure_rm_networkwatcher
+version_added: "2.5.0"
+short_description: Manage the network watcher
+description:
+    - Create, update or delete the network watcher.
+options:
+    resource_group:
+        description:
+            - Name of resource group.
+        required: true
+        type: str
+    location:
+        description:
+            - Valid Azure location. Defaults to location of the resource group.
+        type: str
+    name:
+        description:
+            - The name of the network watcher.
+        required: true
+        type: str
+    state:
+        description:
+            - State of the Network Watcher. Use C(present) to create or update and C(absent) to delete.
+        default: present
+        type: str
+        choices:
+            - absent
+            - present
+
+extends_documentation_fragment:
+    - azure.azcollection.azure
+    - azure.azcollection.azure_tags
+
+author:
+    - xuzhang3 (@xuzhang3)
+    - Fred-sun (@Fred-sun)
+
+'''
+
+EXAMPLES = '''
+- name: Create a Network Watcher
+  azure_rm_networkwatcher:
+    resource_group: myResourceGroup
+    name: myNetworkWatcher
+    location: eastus
+    tags:
+      testing: testing
+      delete: on-exit
+
+- name: Delete a Network Watcher
+  azure_rm_networkwatcher:
+    resource_group: myResourceGroup
+    name: myNetworkWatcher
+    state: absent
+'''
+RETURN = '''
+state:
+    description:
+        - The facts of the network watcher.
+    returned: always
+    type: complex
+    contains:
+        resource_group:
+            description:
+                - The resource group.
+            type: str
+            returned: always
+            sample: NetworkWatcherRG
+        id:
+            description:
+                - Resource ID.
+            returned: always
+            type: str
+            sample: "/subscriptions/xxx-xxx/resourceGroups/NetworkWatcherRG/providers/Microsoft.Network/networkWatchers/netwatcher_eastus"
+        location:
+            description:
+                - Resource location.
+            returned: always
+            type: str
+            sample: eastus
+        name:
+            description:
+                - Resource name.
+            returned: always
+            type: str
+            sample: mynetworkwatcher01
+        tags:
+            description:
+                - Resource tags.
+            returned: always
+            type: dict
+            sample: { 'key1':'value1' }
+        type:
+            description:
+                - Resource type.
+            returned: always
+            type: str
+            sample: "Microsoft.Network/networkWatchers"
+        provisioning_state:
+            description:
+                - The provisioning state of the network watcher resource.
+            type: str
+            returned: always
+            sample: Succeeded
+'''
+
+try:
+    from azure.core.exceptions import ResourceNotFoundError
+except ImportError:
+    # This is handled in azure_rm_common
+    pass
+
+from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common import AzureRMModuleBase
+
+
+class AzureRMNetworkWatcher(AzureRMModuleBase):
+
+    def __init__(self):
+
+        self.module_arg_spec = dict(
+            resource_group=dict(type='str', required=True),
+            name=dict(type='str', required=True),
+            state=dict(type='str', default='present', choices=['present', 'absent']),
+            location=dict(type='str'),
+        )
+
+        self.resource_group = None
+        self.name = None
+        self.state = None
+        self.location = None
+        self.tags = None
+
+        self.results = dict(
+            changed=False,
+            state=dict()
+        )
+
+        super(AzureRMNetworkWatcher, self).__init__(self.module_arg_spec,
+                                                    supports_tags=True,
+                                                    supports_check_mode=True)
+
+    def exec_module(self, **kwargs):
+
+        for key in list(self.module_arg_spec.keys()) + ['tags']:
+            setattr(self, key, kwargs[key])
+
+        resource_group = self.get_resource_group(self.resource_group)
+        if not self.location:
+            # Set default location
+            self.location = resource_group.location
+
+        changed = False
+        results = dict()
+
+        old_response = self.get_by_name()
+
+        if old_response is not None:
+            if self.state == 'present':
+                update_tags, new_tags = self.update_tags(old_response['tags'])
+                if update_tags:
+                    changed = True
+                    if not self.check_mode:
+                        results = self.update_watcher_tags(dict(tags=new_tags))
+                else:
+                    results = old_response
+            else:
+                changed = True
+                if not self.check_mode:
+                    results = self.delete_network_watcher()
+        else:
+            if self.state == 'present':
+                changed = True
+                if not self.check_mode:
+                    results = self.create_or_update(dict(tags=self.tags, location=self.location))
+            else:
+                changed = False
+                self.log("The Network Watcher is not exists")
+
+        self.results['changed'] = changed
+        self.results['state'] = results
+
+        return self.results
+
+    def get_by_name(self):
+        response = None
+        try:
+            response = self.network_client.network_watchers.get(self.resource_group, self.name)
+
+        except ResourceNotFoundError as exec:
+            self.log("Failed to get network watcher, Exception as {0}".format(exec))
+
+        return self.to_dict(response)
+
+    def create_or_update(self, body):
+        response = None
+        try:
+            response = self.to_dict(self.network_client.network_watchers.create_or_update(self.resource_group, self.name, body))
+        except Exception as exc:
+            self.fail("Error creating Network Watcher {0} - {1}".format(self.name, str(exc)))
+
+        return response
+
+    def update_watcher_tags(self, body):
+        response = None
+        try:
+            response = self.network_client.network_watchers.update_tags(self.resource_group, self.name, body)
+        except Exception as exc:
+            self.fail("Error updating Network Watcher {0} - {1}".format(self.name, str(exc)))
+        return self.to_dict(response)
+
+    def delete_network_watcher(self):
+        try:
+            self.network_client.network_watchers.begin_delete(self.resource_group, self.name)
+        except Exception as exc:
+            self.fail("Error deleting Network Watcher {0} - {1}".format(self.name, str(exc)))
+
+    def to_dict(self, body):
+        results = None
+        if body is not None:
+            results = dict(
+                resource_group=self.resource_group,
+                id=body.id,
+                name=body.name,
+                location=body.location,
+                tags=body.tags,
+                type=body.type,
+                provisioning_state=body.provisioning_state,
+            )
+        return results
+
+
+def main():
+    AzureRMNetworkWatcher()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/plugins/modules/azure_rm_networkwatcher_info.py b/plugins/modules/azure_rm_networkwatcher_info.py
new file mode 100644
index 0000000000..4672f5579e
--- /dev/null
+++ b/plugins/modules/azure_rm_networkwatcher_info.py
@@ -0,0 +1,211 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2024 xuzhang3 (@xuzhang3), Fred-sun (@Fred-sun)
+#
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = '''
+---
+module: azure_rm_networkwatcher_info
+
+version_added: "2.5.0"
+
+short_description: Get or list the network watcher facts
+
+description:
+    - Get or list the network watcher facts.
+
+options:
+    resource_group:
+        description:
+            - The name of the resource group.
+        type: str
+    name:
+        description:
+            - Name of the network watcher.
+        type: str
+    tags:
+        description:
+            - Limit results by providing a list of tags. Format tags as 'key' or 'key:value'.
+        type: list
+        elements: str
+
+extends_documentation_fragment:
+    - azure.azcollection.azure
+
+author:
+    - xuzhang3 (@xuzhang3)
+    - Fred-sun (@Fred-sun)
+
+'''
+
+EXAMPLES = '''
+- name: Get the network watcher facts
+  azure_rm_networkwatcher_info:
+    resource_group: myResourceGroup
+    name: mywatcher01
+
+- name: list the network watcher facts
+  azure_rm_networkwatcher_info:
+    resource_group: myResourceGroup
+
+- name: list the network watcher and filter by tags
+  azure_rm_networkwatcher_info:
+    tags:
+      - key1
+      - key2
+'''
+
+RETURN = '''
+network_watchers:
+    description:
+        - The facts of the network watcher.
+    returned: always
+    type: complex
+    contains:
+        resource_group:
+            description:
+                - The resource group.
+            type: str
+            returned: always
+            sample: NetworkWatcherRG
+        id:
+            description:
+                - Resource ID.
+            returned: always
+            type: str
+            sample: "/subscriptions/xxx-xxx/resourceGroups/NetworkWatcherRG/providers/Microsoft.Network/networkWatchers/netwatcher_eastus"
+        location:
+            description:
+                - Resource location.
+            returned: always
+            type: str
+            sample: eastus
+        name:
+            description:
+                - Resource name.
+            returned: always
+            type: str
+            sample: mynetworkwatcher01
+        tags:
+            description:
+                - Resource tags.
+            returned: always
+            type: dict
+            sample: { 'key1':'value1' }
+        type:
+            description:
+                - Resource type.
+            returned: always
+            type: str
+            sample: "Microsoft.Network/networkWatchers"
+        provisioning_state:
+            description:
+                - The provisioning state of the network watcher resource.
+            type: str
+            returned: always
+            sample: Succeeded
+'''
+
+try:
+    from azure.core.exceptions import ResourceNotFoundError
+    from azure.mgmt.core.tools import parse_resource_id
+except Exception:
+    # This is handled in azure_rm_common
+    pass
+
+from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common import AzureRMModuleBase
+
+
+class AzureRMNetworkWatcherInfo(AzureRMModuleBase):
+
+    def __init__(self):
+
+        self.module_arg_spec = dict(
+            resource_group=dict(type='str'),
+            name=dict(type='str'),
+            tags=dict(type='list', elements='str')
+        )
+
+        self.results = dict(
+            changed=False,
+            network_watchers=[]
+        )
+
+        self.resource_group = None
+        self.name = None
+        self.tags = None
+
+        super(AzureRMNetworkWatcherInfo, self).__init__(self.module_arg_spec,
+                                                        supports_check_mode=True,
+                                                        supports_tags=False,
+                                                        facts_module=True)
+
+    def exec_module(self, **kwargs):
+        for key in self.module_arg_spec:
+            setattr(self, key, kwargs[key])
+
+        if self.name and self.resource_group:
+            response = [self.get_by_name()]
+        elif self.resource_group:
+            response = self.list_by_resourcegroup()
+        else:
+            response = self.list_all()
+
+        self.results['network_watchers'] = [self.to_dict(item) for item in response if response is not None]
+
+        return self.results
+
+    def get_by_name(self):
+        response = None
+        try:
+            response = self.network_client.network_watchers.get(self.resource_group, self.name)
+
+        except ResourceNotFoundError as exec:
+            self.log("Failed to get network watchers, Exception as {0}".format(exec))
+
+        return response
+
+    def list_by_resourcegroup(self):
+        response = None
+        try:
+            response = self.network_client.network_watchers.list(self.resource_group)
+        except Exception as exec:
+            self.log("Faild to list network watchers by resource group, exception as {0}".format(exec))
+        return response
+
+    def list_all(self):
+        response = None
+        try:
+            response = self.network_client.network_watchers.list_all()
+        except Exception as exc:
+            self.fail("Failed to list all items - {0}".format(str(exc)))
+
+        return response
+
+    def to_dict(self, body):
+        results = dict()
+        if body is not None and self.has_tags(body.tags, self.tags):
+            results = dict(
+                resource_group=parse_resource_id(body.id).get('resource_group'),
+                id=body.id,
+                name=body.name,
+                location=body.location,
+                tags=body.tags,
+                type=body.type,
+                provisioning_state=body.provisioning_state,
+            )
+            return results
+        return None
+
+
+def main():
+    AzureRMNetworkWatcherInfo()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/pr-pipelines.yml b/pr-pipelines.yml
index 23c9b562ea..1d1b51e936 100644
--- a/pr-pipelines.yml
+++ b/pr-pipelines.yml
@@ -145,6 +145,7 @@ parameters:
       - "azure_rm_sqlmidbshorttermretentionpolicy"
       - "azure_rm_vmssnetworkinterface_info"
       - "azure_rm_sshpublickey"
+      - "azure_rm_networkwatcher"
       - "inventory_azure"
       - "setup_azure"
 
diff --git a/tests/integration/targets/azure_rm_networkwatcher/aliases b/tests/integration/targets/azure_rm_networkwatcher/aliases
new file mode 100644
index 0000000000..9eb4088566
--- /dev/null
+++ b/tests/integration/targets/azure_rm_networkwatcher/aliases
@@ -0,0 +1,4 @@
+cloud/azure
+shippable/azure/group5
+destructive
+disabled
diff --git a/tests/integration/targets/azure_rm_networkwatcher/meta/main.yml b/tests/integration/targets/azure_rm_networkwatcher/meta/main.yml
new file mode 100644
index 0000000000..95e1952f98
--- /dev/null
+++ b/tests/integration/targets/azure_rm_networkwatcher/meta/main.yml
@@ -0,0 +1,2 @@
+dependencies:
+  - setup_azure
diff --git a/tests/integration/targets/azure_rm_networkwatcher/tasks/main.yml b/tests/integration/targets/azure_rm_networkwatcher/tasks/main.yml
new file mode 100644
index 0000000000..2520c2acf1
--- /dev/null
+++ b/tests/integration/targets/azure_rm_networkwatcher/tasks/main.yml
@@ -0,0 +1,78 @@
+- name: Set Random Variables
+  ansible.builtin.set_fact:
+    rpfx: "{{ resource_group | hash('md5') | truncate(20, True, '') }}"
+
+- name: Create a Network Watcher(Checkmode test)
+  azure_rm_networkwatcher:
+    resource_group: "{{ resource_group }}"
+    name: "mynetwacher{{ rpfx }}"
+    location: westus3
+    tags:
+      testing: testing
+      delete: on-exit
+  check_mode: true
+
+- name: Create a new Network Watcher
+  azure_rm_networkwatcher:
+    resource_group: "{{ resource_group }}"
+    name: "mynetwacher{{ rpfx }}"
+    location: westus3
+    tags:
+      testing: testing
+      delete: on-exit
+  register: output
+
+- name: Assert the network watcher is well created
+  ansible.builtin.assert:
+    that:
+      - output.changed
+
+- name: Create a Network Watcher again(Idempotent test)
+  azure_rm_networkwatcher:
+    resource_group: "{{ resource_group }}"
+    name: "mynetwacher{{ rpfx }}"
+    location: westus3
+    tags:
+      testing: testing
+      delete: on-exit
+  register: output
+
+- name: Assert the network watcher is no change
+  ansible.builtin.assert:
+    that:
+      - not output.changed
+
+- name: Get the network watcher facts
+  azure_rm_networkwatcher_info:
+    resource_group: "{{ resource_group }}"
+    name: "mynetwacher{{ rpfx }}"
+  register: output
+
+- name: Assert the network wacher facts
+  ansible.builtin.assert:
+    that:
+      - output.network_watchers[0].tags | length == 2
+
+- name: Get the network watcher facts and filter by tags
+  azure_rm_networkwatcher_info:
+    resource_group: "{{ resource_group }}"
+    tags:
+      - none-key
+  register: output
+
+- name: Assert the network wacher facts
+  ansible.builtin.assert:
+    that:
+      - output.network_watchers[0] == None
+
+- name: Delete the Network Watcher
+  azure_rm_networkwatcher:
+    resource_group: "{{ resource_group }}"
+    name: "mynetwacher{{ rpfx }}"
+    state: absent
+  register: output
+
+- name: Assert the network watcher is well delete
+  ansible.builtin.assert:
+    that:
+      - output.changed