**Description**

Currently when `ECSOperator` is run, the logs of the ECS task are fetched only when the task is done. That's not so convenient, especially when the task takes some good amount of time. In order to understand what's happening with the task, I need to go to Cloudwatch and search for the tasks logs. It would be good to have some parallel process that could fetch ECS task logs from Cloudwatch and make them visible in a realtime. 

**Are you willing to submit a PR?**

I can try, but I need to be guided.

tests/providers/amazon/aws/operators/test_ecs.py
client_mock = mock.Mock()
        self.ecs.client = client_mock
        self.ecs.arn = 'arn'
        client_mock.describe_tasks.return_value = {
            'tasks': [
                {
                    'stoppedReason': 'Host EC2 (instance i-1234567890abcdef) terminated.',
                    "containers": [
                        {
                            "containerArn": "arn:aws:ecs:us-east-1:012345678910:container/e1ed7aac-d9b2-4315-8726-d2432bf11868",  
                            "lastStatus": "RUNNING",
                            "name": "wordpress",
                            "taskArn": "arn:aws:ecs:us-east-1:012345678910:task/d8c67b3c-ac87-4ffe-a847-4785bc3a8b55",  
                        }
                    ],
                    "desiredStatus": "STOPPED",
                    "lastStatus": "STOPPED",
                    "taskArn": "arn:aws:ecs:us-east-1:012345678910:task/d8c67b3c-ac87-4ffe-a847-4785bc3a8b55",  
                    "taskDefinitionArn": "arn:aws:ecs:us-east-1:012345678910:task-definition/hello_world:11",  
                }
            ]
        }
        with pytest.raises(AirflowException) as ctx:
            self.ecs._check_success_task()
        assert "The task was stopped because the host instance terminated:" in str(ctx.value)
        assert "Host EC2 (" in str(ctx.value)
        assert ") terminated" in str(ctx.value)
        client_mock.describe_tasks.assert_called_once_with(cluster='c', tasks=['arn'])
    def test_check_success_task_not_raises(self):