我们从Python开源项目中，提取了以下7个代码示例，用于说明如何使用docker.sock()。
def main(): ''' Register / De-register containers that have CONSUL_SERVICE_PORT env variable defined ''' args = handler_args() setup_logging(args.verbose) # create a docker client object that talks to the local docker daemon cli = docker.Client(base_url='unix://var/run/docker.sock') con = consul.Consul() logger.info("Consul notifier processing {0}".format(args.action)) if args.action == 'stream': stream(cli, con) elif args.action in ['register', 'deregister']: s = Service(cli, con, args.name) s.handle(args.action) else: logger.error("Unknown action {0}".format(args.action)) sys.exit(1)
def __init__(self, project_dir: pathlib.Path, *, docker_sock='unix://var/run/docker.sock', image='docker', env=None): """ Main class that controls the build process, it provides a simple API so we can easily abstract out other ways to build if we don't end up using docker for them all. It could be cool to have a vagrant way. :param project_dir: Project directory path :param docker_sock: Docker socket :param image: Builder image, it must have docker installed in it. Presuming you will be pulling/pushing from private repos, you should have your docker/config.json in the main user's home. """ self.project_dir = project_dir self.base_url = docker_sock self.image = image self.env = env if env is None: self.env = [] self.cli = None self.container_id = None
def _create_container(self): """Creates the build runner container""" # For debug, you can emulate this with: # docker run -it -v /var/run/docker.sock:/var/run/docker.sock -v "${PWD}":/app \ # --workdir /app -e <your envvars> <manifest.image> sh volumes = ['/app', '/var/run/docker.sock'] host_config = self.cli.create_host_config(binds=[ '{}:/app'.format(self.project_dir.absolute()), '/var/run/docker.sock:/var/run/docker.sock', ]) logger.debug('Creating container "%s"', self.image) container = self.cli.create_container(image=self.image, command='sh', detach=True, environment=self.env, stdin_open=True, working_dir='/app', volumes=volumes, host_config=host_config) return container['Id']
def setUp(self): super(DFPLESecret, self).setUp() # docker-flow-proxy-letsencrypt service dfple_image = self.docker_client.images.build( path=os.path.dirname(os.path.abspath(__file__)), tag='robin/docker-flow-proxy-letsencrypt:{}'.format(self.test_name), quiet=False) dfple_service = { 'name': self.proxy_le_service_name, 'image': 'robin/docker-flow-proxy-letsencrypt:{}'.format(self.test_name), 'constraints': ["node.role == manager"], 'env': [ "DF_PROXY_SERVICE_NAME=proxy_{}".format(self.test_name), "DF_SWARM_LISTENER_SERVICE_NAME=swarm_listener_{}".format(self.test_name), "CERTBOT_OPTIONS=--staging", "LOG=debug", ], 'labels': { "com.df.notify": "true", "com.df.distribute": "true", "com.df.servicePath": "/.well-known/acme-challenge", "com.df.port": "8080", }, 'networks': [self.network_name], 'mounts': ['/var/run/docker.sock:/var/run/docker.sock:rw'], } self.dfple_service = self.docker_client.services.create(**dfple_service) self.services.append(self.dfple_service) # wait until proxy_le service has registered routes proxy_le_present_in_config = self.wait_until_found_in_config([self.proxy_le_service_name]) if not proxy_le_present_in_config: self.get_service_logs(self.proxy_le_service_name) self.assertTrue( proxy_le_present_in_config, "docker-flow-proxy-letsencrypt service not registered.")
def setUp(self): super(DFPLEUpdate, self).setUp() # docker-flow-proxy-letsencrypt service dfple_image = self.docker_client.images.build( path=os.path.dirname(os.path.abspath(__file__)), tag='robin/docker-flow-proxy-letsencrypt:{}'.format(self.test_name), quiet=False) dfple_service = { 'name': self.proxy_le_service_name, 'image': 'robin/docker-flow-proxy-letsencrypt:{}'.format(self.test_name), 'constraints': ["node.role == manager"], 'env': [ "DF_PROXY_SERVICE_NAME=proxy_{}".format(self.test_name), "DF_SWARM_LISTENER_SERVICE_NAME=swarm_listener_{}".format(self.test_name), "CERTBOT_OPTIONS=--staging", "LOG=debug", ], 'labels': { "com.df.notify": "true", "com.df.distribute": "true", "com.df.servicePath": "/.well-known/acme-challenge", "com.df.port": "8080", }, 'networks': [self.network_name], 'mounts': ['/var/run/docker.sock:/var/run/docker.sock:rw'], } self.dfple_service = self.docker_client.services.create(**dfple_service) self.services.append(self.dfple_service) # wait until proxy_le service has registered routes proxy_le_present_in_config = self.wait_until_found_in_config([self.proxy_le_service_name]) if not proxy_le_present_in_config: self.get_service_logs(self.proxy_le_service_name) self.assertTrue( proxy_le_present_in_config, "docker-flow-proxy-letsencrypt service not registered.")
def __enter__(self): logger.debug('Using docker sock %s', self.base_url) self.cli = docker.Client(base_url=self.base_url, version='auto') self._pull_container() self.container_id = self._create_container() self._start_container() return self
def get_client(): """ Set DOCKER_HOST (and probably DOCKER_TLS_VERIFY and DOCKER_CERT_PATH) to connect to a docker instance through TCP. Leave DOCKER_HOST unset and it will use the default, typically unix:/var/run/docker.sock It also needs to know how to connect to ports on the docker container after creating it. Set DOCKER_NET_HOST to provide an IP address to connect to the VNC ports on otherwise if DOCKER_HOST has a hostname, it will connect to the VNC ports using that name. otherwise it connects using localhost """ info = {} host = os.environ.get('DOCKER_HOST') net_host = os.environ.get('DOCKER_NET_HOST') client_api_version = os.environ.get('DOCKER_API_VERSION') if not client_api_version: client_api_version = "auto" # IP to use for started containers if net_host: info['host'] = net_host elif host: info['host'] = urlparse.urlparse(host).netloc.split(':')[0] else: info['host'] = 'localhost' verify = os.environ.get('DOCKER_TLS_VERIFY') == '1' if verify: # use TLS assert_hostname = None cert_path = os.environ.get('DOCKER_CERT_PATH') if cert_path: client_cert = (os.path.join(cert_path, 'cert.pem'), os.path.join(cert_path, 'key.pem')) ca_cert = os.path.join(cert_path, 'ca.pem') else: client_cert = ca_cert = None tls_config = docker.tls.TLSConfig( client_cert=client_cert, ca_cert=ca_cert, verify=verify, assert_hostname=assert_hostname, ) return docker.Client(base_url=host, tls=tls_config, version=client_api_version), info else: return docker.Client(base_url=host, version=client_api_version), info
