
**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
dag_ids: Optional[List[str]],
        pickle_dags: bool,
        async_mode: bool = True,
    ):
        super().__init__()
        self._file_paths: List[str] = []
        self._file_path_queue: List[str] = []
        self._dag_directory = dag_directory
        self._max_runs = max_runs
        self._signal_conn = signal_conn
        self._pickle_dags = pickle_dags
        self._dag_ids = dag_ids
        self._async_mode = async_mode
        self._parsing_start_time: Optional[int] = None
        
        
        
        #
        
        
        if self._async_mode:
            os.set_blocking(self._signal_conn.fileno(), False)
        self._parallelism = conf.getint('scheduler', 'parsing_processes')
        if conf.get('core', 'sql_alchemy_conn').startswith('sqlite') and self._parallelism > 1:
            self.log.warning(
                "Because we cannot use more than 1 thread (parsing_processes = "
                "%d ) when using sqlite. So we set parallelism to 1.",
                self._parallelism,
            )
            self._parallelism = 1