"""Utilities for wandb verify.""" import contextlib import getpass import io import os import time from functools import partial from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple, Union import click import requests from wandb_gql import gql import wandb from wandb.sdk.artifacts.artifact import Artifact from wandb.sdk.lib import runid from ...apis.internal import Api PROJECT_NAME = "verify" GET_RUN_MAX_TIME = 10 MIN_RETRYS = 3 CHECKMARK = "\u2705" RED_X = "\u274c" ID_PREFIX = runid.generate_id() def nice_id(name): return ID_PREFIX + "-" + name def print_results( failed_test_or_tests: Optional[Union[str, List[str]]], warning: bool ) -> None: if warning: color = "yellow" else: color = "red" if isinstance(failed_test_or_tests, str): print(RED_X) # noqa: T201 print(click.style(failed_test_or_tests, fg=color, bold=True)) # noqa: T201 elif isinstance(failed_test_or_tests, list) and len(failed_test_or_tests) > 0: print(RED_X) # noqa: T201 print( # noqa: T201 "\n".join( [click.style(f, fg=color, bold=True) for f in failed_test_or_tests] ) ) else: print(CHECKMARK) # noqa: T201 def check_host(host: str) -> bool: if host in ("api.wandb.ai", "http://api.wandb.ai", "https://api.wandb.ai"): print_results("Cannot run wandb verify against api.wandb.ai", False) return False return True def check_logged_in(api: Api, host: str) -> bool: print("Checking if logged in".ljust(72, "."), end="") # noqa: T201 login_doc_url = "https://docs.wandb.ai/ref/cli/wandb-login" fail_string = None if api.api_key is None: fail_string = ( "Not logged in. Please log in using `wandb login`. See the docs: {}".format( click.style(login_doc_url, underline=True, fg="blue") ) ) # check that api key is correct # TODO: Better check for api key is correct else: res = api.api.viewer() if not res: fail_string = ( "Could not get viewer with default API key. " f"Please relogin using `WANDB_BASE_URL={host} wandb login --relogin` and try again" ) print_results(fail_string, False) return fail_string is None def check_secure_requests(url: str, test_url_string: str, failure_output: str) -> None: # check if request is over https print(test_url_string.ljust(72, "."), end="") # noqa: T201 fail_string = None if not url.startswith("https"): fail_string = failure_output print_results(fail_string, True) def check_cors_configuration(url: str, origin: str) -> None: print("Checking CORs configuration of the bucket".ljust(72, "."), end="") # noqa: T201 fail_string = None res_get = requests.options( url, headers={"Origin": origin, "Access-Control-Request-Method": "GET"} ) if res_get.headers.get("Access-Control-Allow-Origin") is None: fail_string = ( "Your object store does not have a valid CORs configuration, " f"you must allow GET and PUT to Origin: {origin}" ) print_results(fail_string, True) def check_run(api: Api) -> bool: print( # noqa: T201 "Checking logged metrics, saving and downloading a file".ljust(72, "."), end="" ) failed_test_strings = [] # set up config n_epochs = 4 string_test = "A test config" dict_test = {"config_val": 2, "config_string": "config string"} list_test = [0, "one", "2"] config = { "epochs": n_epochs, "stringTest": string_test, "dictTest": dict_test, "listTest": list_test, } # create a file to save filepath = "./test with_special-characters.txt" f = open(filepath, "w") f.write("test") f.close() with wandb.init( id=nice_id("check_run"), reinit=True, config=config, project=PROJECT_NAME, ) as run: run_id = run.id entity = run.entity logged = True try: for i in range(1, 11): run.log({"loss": 1.0 / i}, step=i) log_dict = {"val1": 1.0, "val2": 2} run.log({"dict": log_dict}, step=i + 1) except Exception: logged = False failed_test_strings.append( "Failed to log values to run. Contact W&B for support." ) try: run.log({"HT%3ML ": wandb.Html('Link')}) except Exception: failed_test_strings.append( "Failed to log to media. Contact W&B for support." ) run.save(filepath) public_api = wandb.Api() prev_run = public_api.run(f"{entity}/{PROJECT_NAME}/{run_id}") # raise Exception(prev_run.__dict__) if prev_run is None: failed_test_strings.append( "Failed to access run through API. Contact W&B for support." ) print_results(failed_test_strings, False) return False for key, value in config.items(): if prev_run.config.get(key) != value: failed_test_strings.append( "Read config values don't match run config. Contact W&B for support." ) break if logged and ( prev_run.history_keys["keys"]["loss"]["previousValue"] != 0.1 or prev_run.history_keys["lastStep"] != 11 or prev_run.history_keys["keys"]["dict.val1"]["previousValue"] != 1.0 or prev_run.history_keys["keys"]["dict.val2"]["previousValue"] != 2 ): failed_test_strings.append( "History metrics don't match logged values. Check database encoding." ) if logged and prev_run.summary["loss"] != 1.0 / 10: failed_test_strings.append( "Read summary values don't match expected value. Check database encoding, or contact W&B for support." ) # TODO: (kdg) refactor this so it doesn't rely on an exception handler try: read_file = retry_fn(partial(prev_run.file, filepath)) # There's a race where the file hasn't been processed in the queue, # we just retry until we get a download read_file = retry_fn(partial(read_file.download, replace=True)) except Exception: failed_test_strings.append( "Unable to download file. Check SQS configuration, topic configuration and bucket permissions." ) print_results(failed_test_strings, False) return False contents = read_file.read() if contents != "test": failed_test_strings.append( "Contents of downloaded file do not match uploaded contents. Contact W&B for support." ) print_results(failed_test_strings, False) return len(failed_test_strings) == 0 def verify_manifest( downloaded_manifest: Dict[str, Any], computed_manifest: Dict[str, Any], fails_list: List[str], ) -> None: try: for key in computed_manifest.keys(): assert ( computed_manifest[key]["digest"] == downloaded_manifest[key]["digest"] ) assert computed_manifest[key]["size"] == downloaded_manifest[key]["size"] except AssertionError: fails_list.append( "Artifact manifest does not appear as expected. Contact W&B for support." ) def verify_digest( downloaded: "Artifact", computed: "Artifact", fails_list: List[str] ) -> None: if downloaded.digest != computed.digest: fails_list.append( "Artifact digest does not appear as expected. Contact W&B for support." ) def artifact_with_path_or_paths( name: str, verify_dir: Optional[str] = None, singular: bool = False ) -> "Artifact": art = wandb.Artifact(type="artsy", name=name) # internal file with open("verify_int_test.txt", "w") as f: f.write("test 1") f.close() art.add_file(f.name) if singular: return art if verify_dir is None: verify_dir = "./" with art.new_file("verify_a.txt") as f: f.write("test 2") if not os.path.exists(verify_dir): os.makedirs(verify_dir) with open(f"{verify_dir}/verify_1.txt", "w") as f: f.write("1") art.add_dir(verify_dir) file3 = Path(verify_dir) / "verify_3.txt" file3.write_text("3") # reference to local file art.add_reference(file3.resolve().as_uri()) return art def log_use_download_artifact( artifact: "Artifact", alias: str, name: str, download_dir: str, failed_test_strings: List[str], add_extra_file: bool, ) -> Tuple[bool, Optional["Artifact"], List[str]]: with wandb.init( id=nice_id("log_artifact"), reinit=True, project=PROJECT_NAME, config={"test": "artifact log"}, ) as log_art_run: if add_extra_file: with open("verify_2.txt", "w") as f: f.write("2") f.close() artifact.add_file(f.name) try: log_art_run.log_artifact(artifact, aliases=alias) except Exception as e: failed_test_strings.append(f"Unable to log artifact. {e}") return False, None, failed_test_strings with wandb.init( id=nice_id("use_artifact"), project=PROJECT_NAME, config={"test": "artifact use"}, ) as use_art_run: try: used_art = use_art_run.use_artifact(f"{name}:{alias}") except Exception as e: failed_test_strings.append(f"Unable to use artifact. {e}") return False, None, failed_test_strings try: used_art.download(root=download_dir) except Exception: failed_test_strings.append( "Unable to download artifact. Check bucket permissions." ) return False, None, failed_test_strings return True, used_art, failed_test_strings def check_artifacts() -> bool: print("Checking artifact save and download workflows".ljust(72, "."), end="") # noqa: T201 failed_test_strings: List[str] = [] # test checksum sing_art_dir = "./verify_sing_art" alias = "sing_art1" name = nice_id("sing-artys") singular_art = artifact_with_path_or_paths(name, singular=True) cont_test, download_artifact, failed_test_strings = log_use_download_artifact( singular_art, alias, name, sing_art_dir, failed_test_strings, False ) if not cont_test or download_artifact is None: print_results(failed_test_strings, False) return False try: download_artifact.verify(root=sing_art_dir) except ValueError: failed_test_strings.append( "Artifact does not contain expected checksum. Contact W&B for support." ) # test manifest and digest multi_art_dir = "./verify_art" alias = "art1" name = nice_id("my-artys") art1 = artifact_with_path_or_paths(name, "./verify_art_dir", singular=False) cont_test, download_artifact, failed_test_strings = log_use_download_artifact( art1, alias, name, multi_art_dir, failed_test_strings, True ) if not cont_test or download_artifact is None: print_results(failed_test_strings, False) return False if set(os.listdir(multi_art_dir)) != { "verify_a.txt", "verify_2.txt", "verify_1.txt", "verify_3.txt", "verify_int_test.txt", }: failed_test_strings.append( "Artifact directory is missing files. Contact W&B for support." ) computed = wandb.Artifact("computed", type="dataset") computed.add_dir(multi_art_dir) verify_digest(download_artifact, computed, failed_test_strings) computed_manifest = computed.manifest.to_manifest_json()["contents"] downloaded_manifest = download_artifact.manifest.to_manifest_json()["contents"] verify_manifest(downloaded_manifest, computed_manifest, failed_test_strings) print_results(failed_test_strings, False) return len(failed_test_strings) == 0 def check_graphql_put(api: Api, host: str) -> Tuple[bool, Optional[str]]: # check graphql endpoint using an upload print("Checking signed URL upload".ljust(72, "."), end="") # noqa: T201 failed_test_strings = [] gql_fp = "gql_test_file.txt" f = open(gql_fp, "w") f.write("test2") f.close() with wandb.init( id=nice_id("graphql_put"), reinit=True, project=PROJECT_NAME, config={"test": "put to graphql"}, ) as run: run.save(gql_fp) public_api = wandb.Api() prev_run = public_api.run(f"{run.entity}/{PROJECT_NAME}/{run.id}") if prev_run is None: failed_test_strings.append( "Unable to access previous run through public API. Contact W&B for support." ) print_results(failed_test_strings, False) return False, None # TODO: (kdg) refactor this so it doesn't rely on an exception handler try: read_file = retry_fn(partial(prev_run.file, gql_fp)) url = read_file.url read_file = retry_fn(partial(read_file.download, replace=True)) except Exception: failed_test_strings.append( "Unable to read file successfully saved through a put request. Check SQS configurations, bucket permissions and topic configs." ) print_results(failed_test_strings, False) return False, None contents = read_file.read() try: assert contents == "test2" except AssertionError: failed_test_strings.append( "Read file contents do not match saved file contents. Contact W&B for support." ) print_results(failed_test_strings, False) return len(failed_test_strings) == 0, url def check_large_post() -> bool: print( # noqa: T201 "Checking ability to send large payloads through proxy".ljust(72, "."), end="" ) descy = "a" * int(10**7) username = getpass.getuser() failed_test_strings = [] query = gql( """ query Project($entity: String!, $name: String!, $runName: String!, $desc: String!){ project(entityName: $entity, name: $name) { run(name: $runName, desc: $desc) { name summaryMetrics } } } """ ) public_api = wandb.Api() client = public_api._base_client try: client._get_result( query, variable_values={ "entity": username, "name": PROJECT_NAME, "runName": "", "desc": descy, }, timeout=60, ) except Exception as e: if ( isinstance(e, requests.HTTPError) and e.response is not None and e.response.status_code == 413 ): failed_test_strings.append( 'Failed to send a large payload. Check nginx.ingress.kubernetes.io/proxy-body-size is "0".' ) else: failed_test_strings.append( f"Failed to send a large payload with error: {e}." ) print_results(failed_test_strings, False) return len(failed_test_strings) == 0 def check_wandb_version(api: Api) -> None: print("Checking wandb package version is up to date".ljust(72, "."), end="") # noqa: T201 _, server_info = api.viewer_server_info() fail_string = None warning = False max_cli_version = server_info.get("cliVersionInfo", {}).get("max_cli_version", None) min_cli_version = server_info.get("cliVersionInfo", {}).get( "min_cli_version", "0.0.1" ) from packaging.version import parse if parse(wandb.__version__) < parse(min_cli_version): fail_string = f"wandb version out of date, please run pip install --upgrade wandb=={max_cli_version}" elif parse(wandb.__version__) > parse(max_cli_version): fail_string = ( "wandb version is not supported by your local installation. This could " "cause some issues. If you're having problems try: please run `pip " f"install --upgrade wandb=={max_cli_version}`" ) warning = True print_results(fail_string, warning) def check_sweeps(api: Api) -> bool: print("Checking sweep creation and agent execution".ljust(72, "."), end="") # noqa: T201 failed_test_strings: List[str] = [] sweep_config = { "method": "random", "metric": {"goal": "minimize", "name": "score"}, "parameters": { "x": {"values": [0.01, 0.05, 0.1]}, "y": {"values": [1, 2, 3]}, }, "name": "verify_sweep", } try: with contextlib.redirect_stdout(io.StringIO()): sweep_id = wandb.sweep( sweep=sweep_config, project=PROJECT_NAME, entity=api.default_entity ) except Exception as e: failed_test_strings.append(f"Failed to create sweep: {e}") print_results(failed_test_strings, False) return False if not sweep_id: failed_test_strings.append("Sweep creation returned an invalid ID.") print_results(failed_test_strings, False) return False try: def objective(config): score = config.x**3 + config.y return score def main(): with wandb.init(project=PROJECT_NAME) as run: score = objective(run.config) run.log({"score": score}) wandb.agent(sweep_id, function=main, count=10) except Exception as e: failed_test_strings.append(f"Failed to run sweep agent: {e}") print_results(failed_test_strings, False) return False print_results(failed_test_strings, False) return len(failed_test_strings) == 0 def retry_fn(fn: Callable) -> Any: ini_time = time.time() res = None i = 0 while i < MIN_RETRYS or time.time() - ini_time < GET_RUN_MAX_TIME: i += 1 try: res = fn() break except Exception: time.sleep(1) continue return res