def create_cluster(self, cluster, project_id=None, retry=DEFAULT, timeout=DEFAULT):
if isinstance(cluster, dict):<EOL><INDENT>cluster_proto = Cluster()<EOL>cluster = self._dict_to_proto(py_dict=cluster, proto=cluster_proto)<EOL><DEDENT>elif not isinstance(cluster, Cluster):<EOL><INDENT>raise AirflowException(<EOL>"<STR_LIT>")<EOL><DEDENT>self._append_label(cluster, '<STR_LIT>', '<STR_LIT:v>' + version.version)<EOL>self.log.info(<EOL>"<STR_LIT>",<EOL>self.project_id, self.location, cluster.name<EOL>)<EOL>try:<EOL><INDENT>op = self.get_client().create_cluster(project_id=project_id or self.project_id,<EOL>zone=self.location,<EOL>cluster=cluster,<EOL>retry=retry,<EOL>timeout=timeout)<EOL>op = self.wait_for_operation(op)<EOL>return op.target_link<EOL><DEDENT>except AlreadyExists as error:<EOL><INDENT>self.log.info('<STR_LIT>', error.message)<EOL>return self.get_cluster(name=cluster.name).self_link<EOL><DEDENT>
Creates a cluster, consisting of the specified number and type of Google Compute
Engine instances.

:param cluster: A Cluster protobuf or dict. If dict is provided, it must
    be of the same form as the protobuf message
    :class:`google.cloud.container_v1.types.Cluster`
:type cluster: dict or google.cloud.container_v1.types.Cluster
:param project_id: Google Cloud Platform project ID
:type project_id: str
:param retry: A retry object (``google.api_core.retry.Retry``) used to
    retry requests.
    If None is specified, requests will not be retried.
:type retry: google.api_core.retry.Retry
:param timeout: The amount of time, in seconds, to wait for the request to
    complete. Note that if retry is specified, the timeout applies to each
    individual attempt.
:type timeout: float
:return: The full url to the new, or existing, cluster
:raises:
    ParseError: On JSON parsing problems when trying to convert dict
    AirflowException: cluster is not dict type nor Cluster proto type