func getTransformFromResourceData()

in internal/elasticsearch/transform/transform.go [472:640]
146 lines of code
52 McCabe index (conditional complexity)

func getTransformFromResourceData(ctx context.Context, d *schema.ResourceData, name string, serverVersion *version.Version) (*models.Transform, error) {

	var transform models.Transform
	transform.Name = name

	if v, ok := d.GetOk("description"); ok {
		transform.Description = v.(string)
	}

	if v, ok := d.GetOk("source"); ok {
		definedSource := v.([]interface{})[0].(map[string]interface{})

		transform.Source = new(models.TransformSource)
		indices := make([]string, 0)
		for _, i := range definedSource["indices"].([]interface{}) {
			indices = append(indices, i.(string))
		}
		transform.Source.Indices = indices

		if v, ok := definedSource["query"]; ok && len(v.(string)) > 0 {
			var query interface{}
			if err := json.NewDecoder(strings.NewReader(v.(string))).Decode(&query); err != nil {
				return nil, err
			}
			transform.Source.Query = query
		}

		if v, ok := definedSource["runtime_mappings"]; ok && len(v.(string)) > 0 && isSettingAllowed(ctx, "source.runtime_mappings", serverVersion) {
			var runtimeMappings interface{}
			if err := json.NewDecoder(strings.NewReader(v.(string))).Decode(&runtimeMappings); err != nil {
				return nil, err
			}
			transform.Source.RuntimeMappings = runtimeMappings
		}
	}

	if v, ok := d.GetOk("destination"); ok {
		definedDestination := v.([]interface{})[0].(map[string]interface{})

		transform.Destination = &models.TransformDestination{
			Index: definedDestination["index"].(string),
		}

		if aliases, ok := definedDestination["aliases"].([]interface{}); ok && len(aliases) > 0 && isSettingAllowed(ctx, "destination.aliases", serverVersion) {
			transform.Destination.Aliases = make([]models.TransformAlias, len(aliases))
			for i, alias := range aliases {
				aliasMap := alias.(map[string]interface{})
				transform.Destination.Aliases[i] = models.TransformAlias{
					Alias:          aliasMap["alias"].(string),
					MoveOnCreation: aliasMap["move_on_creation"].(bool),
				}
			}
		}

		if pipeline, ok := definedDestination["pipeline"]; ok && isSettingAllowed(ctx, "destination.pipeline", serverVersion) {
			transform.Destination.Pipeline = pipeline.(string)
		}
	}

	if v, ok := d.GetOk("pivot"); ok {
		var pivot interface{}
		if err := json.NewDecoder(strings.NewReader(v.(string))).Decode(&pivot); err != nil {
			return nil, err
		}
		transform.Pivot = pivot
	}

	if v, ok := d.GetOk("latest"); ok && isSettingAllowed(ctx, "latest", serverVersion) {
		var latest interface{}
		if err := json.NewDecoder(strings.NewReader(v.(string))).Decode(&latest); err != nil {
			return nil, err
		}
		transform.Latest = latest
	}

	if v, ok := d.GetOk("frequency"); ok && isSettingAllowed(ctx, "frequency", serverVersion) {
		transform.Frequency = v.(string)
	}

	if v, ok := d.GetOk("metadata"); ok && isSettingAllowed(ctx, "metadata", serverVersion) {
		var metadata map[string]interface{}
		if err := json.NewDecoder(strings.NewReader(v.(string))).Decode(&metadata); err != nil {
			return nil, err
		}
		transform.Meta = metadata
	}

	if v, ok := d.GetOk("retention_policy"); ok && isSettingAllowed(ctx, "retention_policy", serverVersion) {
		definedRetentionPolicy := v.([]interface{})[0].(map[string]interface{})

		if v, ok := definedRetentionPolicy["time"]; ok {
			retentionTime := models.TransformRetentionPolicyTime{}
			var definedRetentionTime = v.([]interface{})[0].(map[string]interface{})
			if f, ok := definedRetentionTime["field"]; ok {
				retentionTime.Field = f.(string)
			}
			if ma, ok := definedRetentionTime["max_age"]; ok {
				retentionTime.MaxAge = ma.(string)
			}
			transform.RetentionPolicy = &models.TransformRetentionPolicy{
				Time: retentionTime,
			}
		}
	}

	if v, ok := d.GetOk("sync"); ok {
		definedSync := v.([]interface{})[0].(map[string]interface{})

		if v, ok := definedSync["time"]; ok {
			syncTime := models.TransformSyncTime{}
			var definedSyncTime = v.([]interface{})[0].(map[string]interface{})
			if f, ok := definedSyncTime["field"]; ok {
				syncTime.Field = f.(string)
			}
			if d, ok := definedSyncTime["delay"]; ok {
				syncTime.Delay = d.(string)
			}
			transform.Sync = &models.TransformSync{
				Time: syncTime,
			}
		}
	}

	// settings
	settings := models.TransformSettings{}
	setSettings := false

	if v, ok := d.GetOk("align_checkpoints"); ok && isSettingAllowed(ctx, "align_checkpoints", serverVersion) {
		setSettings = true
		ac := v.(bool)
		settings.AlignCheckpoints = &ac
	}
	if v, ok := d.GetOk("dates_as_epoch_millis"); ok && isSettingAllowed(ctx, "dates_as_epoch_millis", serverVersion) {
		setSettings = true
		dem := v.(bool)
		settings.DatesAsEpochMillis = &dem
	}
	if v, ok := d.GetOk("deduce_mappings"); ok && isSettingAllowed(ctx, "deduce_mappings", serverVersion) {
		setSettings = true
		dm := v.(bool)
		settings.DeduceMappings = &dm
	}
	if v, ok := d.GetOk("docs_per_second"); ok && isSettingAllowed(ctx, "docs_per_second", serverVersion) {
		setSettings = true
		dps := v.(float64)
		settings.DocsPerSecond = &dps
	}
	if v, ok := d.GetOk("max_page_search_size"); ok && isSettingAllowed(ctx, "max_page_search_size", serverVersion) {
		setSettings = true
		mpss := v.(int)
		settings.MaxPageSearchSize = &mpss
	}
	if v, ok := d.GetOk("num_failure_retries"); ok && isSettingAllowed(ctx, "num_failure_retries", serverVersion) {
		setSettings = true
		nfr := v.(int)
		settings.NumFailureRetries = &nfr
	}
	if v, ok := d.GetOk("unattended"); ok && isSettingAllowed(ctx, "unattended", serverVersion) {
		setSettings = true
		u := v.(bool)
		settings.Unattended = &u
	}

	if setSettings {
		transform.Settings = &settings
	}

	return &transform, nil
}