
**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.



airflow/dag_processing/manager.py
"""Refresh file paths from dag dir if we haven't done it for too long."""
        now = timezone.utcnow()
        elapsed_time_since_refresh = (now - self.last_dag_dir_refresh_time).total_seconds()
        if elapsed_time_since_refresh > self.dag_dir_list_interval:
            
            self.log.info("Searching for files in %s", self._dag_directory)
            self._file_paths = list_py_file_paths(self._dag_directory)
            self.last_dag_dir_refresh_time = now
            self.log.info("There are %s files in %s", len(self._file_paths), self._dag_directory)
            self.set_file_paths(self._file_paths)
            try:
                self.log.debug("Removing old import errors")
                self.clear_nonexistent_import_errors()
            except Exception:
                self.log.exception("Error removing old import errors")
            SerializedDagModel.remove_deleted_dags(self._file_paths)
            DagModel.deactivate_deleted_dags(self._file_paths)
            from airflow.models.dagcode import DagCode
            DagCode.remove_deleted_code(self._file_paths)
    def _print_stat(self):
        """Occasionally print out stats about how fast the files are getting processed"""
        if 0 < self.print_stats_interval < time.monotonic() - self.last_stat_print_time:
            if self._file_paths:
                self._log_file_processing_stats(self._file_paths)
            self.last_stat_print_time = time.monotonic()
    @provide_session
    def clear_nonexistent_import_errors(self, session):