
**Apache Airflow version**: 2.1.2

**Apache Airflow Provider versions** (please include all providers that are relevant to your bug): 2.0

apache-airflow-providers-apache-hive==2.0.0
apache-airflow-providers-celery==2.0.0
apache-airflow-providers-cncf-kubernetes==1.2.0
apache-airflow-providers-docker==2.0.0
apache-airflow-providers-elasticsearch==1.0.4
apache-airflow-providers-ftp==2.0.0
apache-airflow-providers-imap==2.0.0
apache-airflow-providers-jdbc==2.0.0
apache-airflow-providers-microsoft-mssql==2.0.0
apache-airflow-providers-mysql==2.0.0
apache-airflow-providers-oracle==2.0.0
apache-airflow-providers-postgres==2.0.0
apache-airflow-providers-sqlite==2.0.0
apache-airflow-providers-ssh==2.0.0

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): 1.17

**Environment**:

- **Install tools**: https://github.com/prometheus/statsd_exporter

**What happened**:

We are using an [statsd_exporter](https://github.com/prometheus/statsd_exporter) to store statsd metrics in prometheus. And we came across strange behavior, the metric dag_processing.last_duration. <dag_file> for different dags is drawn with holes at random intervals.

![image](https://user-images.githubusercontent.com/3032319/128734876-800ab58e-50d5-4c81-af4a-79c8660c2f0a.png)

![image](https://user-images.githubusercontent.com/3032319/128734946-27381da5-40cc-4c83-9bcc-60d8264d26a3.png)

![image](https://user-images.githubusercontent.com/3032319/128735019-a523e58c-5751-466d-abb7-1888fa79b9e1.png)


**What you expected to happen**:

Metrics should be sent with the frequency specified in the config in the AIRFLOV__SHEDULER__PRINT_STATS_INTERVAL parameter and the default value is 30, and this happens in the[ _log_file_processing_stats method](https://github.com/apache/airflow/blob/2fea4cdceaa12b3ac13f24eeb383af624aacb2e7/airflow/dag_processing/manager.py#L696), the problem is that the initial time is taken using the [get_start_time](https://github.com/apache/airflow/blob/2fea4cdceaa12b3ac13f24eeb383af624aacb2e7/airflow/dag_processing/manager.py#L827) function which looks only at the probability of active processes that some of the dags will have time to be processed in 30 seconds and removed from self._processors [file_path] and thus the metrics for them will not be sent to the statsd. While for outputting to the log,  [lasttime](https://github.com/apache/airflow/blob/2fea4cdceaa12b3ac13f24eeb383af624aacb2e7/airflow/dag_processing/manager.py#L783) is used in which information on the last processing is stored and from which it would be nice to send metrics to the statistics.



tests/dag_processing/test_manager.py
manager.prepare_file_path_queue()
            assert manager._file_path_queue == []
        
        
        file_1_new_mtime = freezed_base_time - timedelta(seconds=5)
        file_1_new_mtime_ts = file_1_new_mtime.timestamp()
        with freeze_time(freezed_base_time):
            manager.set_file_paths(dag_files)
            assert manager._file_path_queue == []
            
            mock_getmtime.side_effect = [file_1_new_mtime_ts]
            manager.prepare_file_path_queue()
            
            assert manager._file_path_queue == ["file_1.py"]
            assert last_finish_time < file_1_new_mtime
            assert (
                manager._file_process_interval
                > (freezed_base_time - manager.get_last_finish_time("file_1.py")).total_seconds()
            )
    def test_find_zombies(self):
        manager = DagFileProcessorManager(
            dag_directory='directory',
            max_runs=1,
            processor_timeout=timedelta.max,
            signal_conn=MagicMock(),
            dag_ids=[],
            pickle_dags=False,
            async_mode=True,
        )
        dagbag = DagBag(TEST_DAG_FOLDER, read_dags_from_db=False)