
**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
pickle_dags=False,
            async_mode=True,
        )
        manager._run_parsing_loop()
        result = None
        while parent_pipe.poll(timeout=None):
            result = parent_pipe.recv()
            if isinstance(result, DagParsingStat) and result.done:
                break
        
        assert sum(stat.run_count for stat in manager._file_stats.values()) == 3
        with create_session() as session:
            assert session.query(DagModel).get(dag_id) is not None
    @conf_vars({('core', 'load_examples'): 'False'})
    @pytest.mark.backend("mysql", "postgres")
    @pytest.mark.execution_timeout(30)
    @mock.patch('airflow.dag_processing.manager.DagFileProcessorProcess')
    def test_pipe_full_deadlock(self, mock_processor):
        dag_filepath = TEST_DAG_FOLDER / "test_scheduler_dags.py"
        child_pipe, parent_pipe = multiprocessing.Pipe()
        
        for fd in (parent_pipe.fileno(),):
            sock = socket.socket(fileno=fd)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
            sock.detach()
        exit_event = threading.Event()
        
        
        def keep_pipe_full(pipe, exit_event):