Refactor docker info fetches to be async

Refactor docker metadata fetches to be asynchronous, using the framework
in sysdig::async_key_value_source.

docker_async_source (a global static so it can be long-lived) is now
responsible for looking up docker metadata. Some methods that used to be
in docker:: like parse_docker(), get_docker(), etc move to
docker_async_source. It dequeues requests, calling parse_docker() to
look up the information as needed, and calls store_value() to pass the
metadata back to the caller.

docker::resolve now uses parse_docker_async to schedule the lookup with
docker_async_source. Before scheduling the lookup it creates a stub
container with type UNKNOWN (UNKNOWN because you can't tell the
difference between cri and docker containers only from the cgroup), with
the id set, and with a name/image set to "incomplete". This ensures that
threads have some associated container info object with it. In the
callback once the full metadata is available, it calls
notify_new_container, which creates the CONTAINER_JSON event and pushes
it to the inspector.

There's a fair amount of bookeeping to make sure that the container
metadata has a valid tid. The very first tid that creates the container
often exits after forking of the real container entrypoint, so you need
to keep track of the "top tid" in the container for every call to
docker::resolve() and replace it if you find it's exited.

userspace/libsinsp/CMakeLists.txt

@@ -18,6 +18,7 @@
 include_directories(./)
 include_directories(../../common)
 include_directories(../libscap)
+include_directories(../async)
 include_directories("${JSONCPP_INCLUDE}")
 include_directories("${LUAJIT_INCLUDE}")
 

userspace/libsinsp/container.cpp

@@ -413,7 +413,7 @@ void sinsp_container_manager::cleanup()
 
 void sinsp_container_manager::set_query_docker_image_info(bool query_image_info)
 {
-	libsinsp::container_engine::docker::set_query_image_info(query_image_info);
+	libsinsp::container_engine::docker_async_source::set_query_image_info(query_image_info);
 }
 
 void sinsp_container_manager::set_cri_extra_queries(bool extra_queries)

userspace/libsinsp/container_engine/docker.h

@@ -22,9 +22,16 @@ limitations under the License.
 #include <memory>
 #include <string>
 #include <vector>
+#include <atomic>
+
+#include "tbb/concurrent_hash_map.h"
 
 #include "json/json.h"
 
+#include "async_key_value_source.h"
+
+#include "sinsp.h"
+
 #include "container_info.h"
 
 class sinsp_container_manager;
@@ -33,7 +40,8 @@ class sinsp_threadinfo;
 
 namespace libsinsp {
 namespace container_engine {
-class docker
+
+class docker_async_source : public sysdig::async_key_value_source<std::string, sinsp_container_info>
 {
 	enum docker_response
 	{
@@ -43,20 +51,64 @@ class docker
 	};
 
 public:
-	docker();
+	docker_async_source(uint64_t max_wait_ms, uint64_t ttl_ms);
+	virtual ~docker_async_source();
 
-	bool resolve(sinsp_container_manager* manager, sinsp_threadinfo* tinfo, bool query_os_for_missing_info);
-	static void cleanup();
+	void set_inspector(sinsp *inspector);
 	static void set_query_image_info(bool query_image_info);
-	static void parse_json_mounts(const Json::Value &mnt_obj, std::vector<sinsp_container_info::container_mount_info> &mounts);
+
+	// Note that this tid is the current top tid for this container
+	void set_top_tid(const std::string &container_id, sinsp_threadinfo *tinfo);
+
+	// Update the mapping from container id to top running tid for
+	// that container.
+	void update_top_tid(std::string &container_id, sinsp_threadinfo *tinfo);
+
+	// Get the thread id of the top thread running in this container.
+	int64_t get_top_tid(const std::string &container_id);
+
+        bool pending_lookup(std::string &container_id);
 
 protected:
-	docker_response get_docker(sinsp_container_manager* manager, const std::string& url, std::string &json);
+	void run_impl();
+
 	std::string build_request(const std::string& url);
-	bool parse_docker(sinsp_container_manager* manager, sinsp_container_info *container, sinsp_threadinfo* tinfo);
+
+	docker_response get_docker(const std::string& url, std::string &json);
+	bool parse_docker(std::string &container_id, sinsp_container_info *container);
 
 	static std::string m_api_version;
+	sinsp *m_inspector;
+
 	static bool m_query_image_info;
+
+	// Maps from container id to the "top" threadinfo in the
+	// process heirarchy having that container id that
+	// exists. These associations are only maintained while an
+	// async lookup of container information is in progress. We
+	// use this to ensure that the tid of the CONTAINER_JSON event
+	// we eventually emit is the top running thread in the
+	// container.
+	typedef tbb::concurrent_hash_map<std::string, int64_t> top_tid_table;
+	top_tid_table m_top_tids;
+};
+
+class docker
+{
+public:
+	docker();
+
+	bool resolve(sinsp_container_manager* manager, sinsp_threadinfo* tinfo, bool query_os_for_missing_info);
+	static void cleanup();
+	static void parse_json_mounts(const Json::Value &mnt_obj, std::vector<sinsp_container_info::container_mount_info> &mounts);
+	static void set_enabled(bool enabled);
+
+	// Container name only set for windows. For linux name must be fetched via lookup
+	static bool detect_docker(const sinsp_threadinfo* tinfo, std::string& container_id, std::string &container_name);
+protected:
+	void parse_docker_async(sinsp *inspector, std::string &container_id, sinsp_container_manager *manager);
+
+	static std::unique_ptr<docker_async_source> g_docker_info_source;
 	static bool m_enabled;
 };
 }