Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

DG-1476 Handle QN check and Archive Stakeholders in domain delete flow #3326

Open
wants to merge 21 commits into
base: master
Choose a base branch
from
Open

DG-1476 Handle QN check and Archive Stakeholders in domain delete flow #3326

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
2378f67
fix QN check in datadomain and dataproduct create request flow
akshaysw May 21, 2024
c5dd93b
revert line change
akshaysw May 23, 2024
ea92d4e
merge master
akshaysw May 23, 2024
5238a6b
add check for stakeholder
akshaysw Jun 5, 2024
03dc646
add check for product and domain update flow
akshaysw Jun 14, 2024
a971da0
handle review comments
akshaysw Jun 14, 2024
756a05e
handle review comments
akshaysw Jun 14, 2024
5653f9e
handle move domain case
akshaysw Jun 17, 2024
c743860
logic to delete stakeholders before archiving domain
PRATHAM2002-DS Jul 9, 2024
835ecf0
resolved conflicts
PRATHAM2002-DS Jul 9, 2024
36c589c
corrected indexsearch dsl
PRATHAM2002-DS Jul 10, 2024
11247f3
validation for stakeholderTitle
PRATHAM2002-DS Jul 15, 2024
6532187
added validation
PRATHAM2002-DS Jul 15, 2024
c21b8d7
corrected if logic
PRATHAM2002-DS Jul 15, 2024
26a45a3
throw error only when Domain qn is removed
PRATHAM2002-DS Jul 15, 2024
8f41ce4
changed error message
PRATHAM2002-DS Jul 15, 2024
8301c3c
optimised validation logic
PRATHAM2002-DS Jul 16, 2024
32ecfbc
handled all domains testcase
PRATHAM2002-DS Jul 16, 2024
ab8a5dd
resolved conflicts
PRATHAM2002-DS Jul 16, 2024
3827bed
resolved conflicts
PRATHAM2002-DS Jul 16, 2024
fea7d5d
skip deleting Titles with active stakeholders
PRATHAM2002-DS Jul 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1892,7 +1892,7 @@ public static String getDelimitedClassificationNames(Collection<String> classifi
* @return Iterator of children vertices
*/
public static Iterator<AtlasVertex> getActiveParentVertices(AtlasVertex vertex, String parentEdgeLabel) throws AtlasBaseException {
return getActiveVertices(vertex, parentEdgeLabel, AtlasEdgeDirection.IN);
return getActiveVertices(vertex, AtlasEdgeDirection.IN, parentEdgeLabel);
}

/**
Expand All @@ -1901,11 +1901,12 @@ public static Iterator<AtlasVertex> getActiveParentVertices(AtlasVertex vertex,
* @param childrenEdgeLabel Edge label of children
* @return Iterator of children vertices
*/
public static Iterator<AtlasVertex> getActiveChildrenVertices(AtlasVertex vertex, String childrenEdgeLabel) throws AtlasBaseException {
return getActiveVertices(vertex, childrenEdgeLabel, AtlasEdgeDirection.OUT);

public static Iterator<AtlasVertex> getActiveChildrenVertices(AtlasVertex vertex, String... childrenEdgeLabel) throws AtlasBaseException {
return getActiveVertices(vertex, AtlasEdgeDirection.OUT, childrenEdgeLabel);
}

public static Iterator<AtlasVertex> getActiveVertices(AtlasVertex vertex, String childrenEdgeLabel, AtlasEdgeDirection direction) throws AtlasBaseException {
public static Iterator<AtlasVertex> getActiveVertices(AtlasVertex vertex, AtlasEdgeDirection direction, String... childrenEdgeLabel) throws AtlasBaseException {
AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("CategoryPreProcessor.getEdges");

try {
Expand Down
30 changes: 16 additions & 14 deletions repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1518,21 +1518,23 @@ private EntityMutationResponse createOrUpdate(EntityStream entityStream, boolean
// Check if authorized to update entities
if (!reqContext.isImportInProgress()) {
for (AtlasEntity entity : context.getUpdatedEntities()) {
AtlasEntityHeader entityHeaderWithClassifications = entityRetriever.toAtlasEntityHeaderWithClassifications(entity.getGuid());
AtlasEntityHeader entityHeader = new AtlasEntityHeader(entity);
if(!PreProcessor.skipUpdateAuthCheckTypes.contains(entity.getTypeName())){
AtlasEntityHeader entityHeaderWithClassifications = entityRetriever.toAtlasEntityHeaderWithClassifications(entity.getGuid());
AtlasEntityHeader entityHeader = new AtlasEntityHeader(entity);

if(CollectionUtils.isNotEmpty(entityHeaderWithClassifications.getClassifications())) {
entityHeader.setClassifications(entityHeaderWithClassifications.getClassifications());
}
if(CollectionUtils.isNotEmpty(entityHeaderWithClassifications.getClassifications())) {
entityHeader.setClassifications(entityHeaderWithClassifications.getClassifications());
}

AtlasEntity diffEntity = reqContext.getDifferentialEntity(entity.getGuid());
boolean skipAuthBaseConditions = diffEntity != null && MapUtils.isEmpty(diffEntity.getCustomAttributes()) && MapUtils.isEmpty(diffEntity.getBusinessAttributes()) && CollectionUtils.isEmpty(diffEntity.getClassifications()) && CollectionUtils.isEmpty(diffEntity.getLabels());
boolean skipAuthMeaningsUpdate = diffEntity != null && MapUtils.isNotEmpty(diffEntity.getRelationshipAttributes()) && diffEntity.getRelationshipAttributes().containsKey("meanings") && diffEntity.getRelationshipAttributes().size() == 1 && MapUtils.isEmpty(diffEntity.getAttributes());
boolean skipAuthStarredDetailsUpdate = diffEntity != null && MapUtils.isEmpty(diffEntity.getRelationshipAttributes()) && MapUtils.isNotEmpty(diffEntity.getAttributes()) && diffEntity.getAttributes().size() == 3 && diffEntity.getAttributes().containsKey(ATTR_STARRED_BY) && diffEntity.getAttributes().containsKey(ATTR_STARRED_COUNT) && diffEntity.getAttributes().containsKey(ATTR_STARRED_DETAILS_LIST);
if (skipAuthBaseConditions && (skipAuthMeaningsUpdate || skipAuthStarredDetailsUpdate)) {
//do nothing, only diff is relationshipAttributes.meanings or starred, allow update
} else {
AtlasAuthorizationUtils.verifyUpdateEntityAccess(typeRegistry, entityHeader,"update entity: type=" + entity.getTypeName());
AtlasEntity diffEntity = reqContext.getDifferentialEntity(entity.getGuid());
boolean skipAuthBaseConditions = diffEntity != null && MapUtils.isEmpty(diffEntity.getCustomAttributes()) && MapUtils.isEmpty(diffEntity.getBusinessAttributes()) && CollectionUtils.isEmpty(diffEntity.getClassifications()) && CollectionUtils.isEmpty(diffEntity.getLabels());
boolean skipAuthMeaningsUpdate = diffEntity != null && MapUtils.isNotEmpty(diffEntity.getRelationshipAttributes()) && diffEntity.getRelationshipAttributes().containsKey("meanings") && diffEntity.getRelationshipAttributes().size() == 1 && MapUtils.isEmpty(diffEntity.getAttributes());
boolean skipAuthStarredDetailsUpdate = diffEntity != null && MapUtils.isEmpty(diffEntity.getRelationshipAttributes()) && MapUtils.isNotEmpty(diffEntity.getAttributes()) && diffEntity.getAttributes().size() == 3 && diffEntity.getAttributes().containsKey(ATTR_STARRED_BY) && diffEntity.getAttributes().containsKey(ATTR_STARRED_COUNT) && diffEntity.getAttributes().containsKey(ATTR_STARRED_DETAILS_LIST);
if (skipAuthBaseConditions && (skipAuthMeaningsUpdate || skipAuthStarredDetailsUpdate)) {
//do nothing, only diff is relationshipAttributes.meanings or starred, allow update
} else {
AtlasAuthorizationUtils.verifyUpdateEntityAccess(typeRegistry, entityHeader,"update entity: type=" + entity.getTypeName());
}
}
}
}
Expand Down Expand Up @@ -1832,7 +1834,7 @@ public PreProcessor getPreProcessor(String typeName) {
break;

case DATA_DOMAIN_ENTITY_TYPE:
preProcessor = new DataDomainPreProcessor(typeRegistry, entityRetriever, graph);
preProcessor = new DataDomainPreProcessor(typeRegistry, entityRetriever, graph, this);
break;

case DATA_PRODUCT_ENTITY_TYPE:
Expand Down
12 changes: 8 additions & 4 deletions ...y/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/PreProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,7 @@
import java.util.HashSet;
import java.util.Set;

import static org.apache.atlas.repository.Constants.ATLAS_GLOSSARY_CATEGORY_ENTITY_TYPE;
import static org.apache.atlas.repository.Constants.ATLAS_GLOSSARY_TERM_ENTITY_TYPE;
import static org.apache.atlas.repository.Constants.STAKEHOLDER_ENTITY_TYPE;
import static org.apache.atlas.repository.Constants.STAKEHOLDER_TITLE_ENTITY_TYPE;
import static org.apache.atlas.repository.Constants.*;


public interface PreProcessor {
Expand All @@ -22,6 +19,13 @@ public interface PreProcessor {
add(ATLAS_GLOSSARY_CATEGORY_ENTITY_TYPE);
add(STAKEHOLDER_ENTITY_TYPE);
add(STAKEHOLDER_TITLE_ENTITY_TYPE);
add(DATA_DOMAIN_ENTITY_TYPE);
add(DATA_PRODUCT_ENTITY_TYPE);
}};

Set<String> skipUpdateAuthCheckTypes = new HashSet<String>() {{
add(DATA_DOMAIN_ENTITY_TYPE);
add(DATA_PRODUCT_ENTITY_TYPE);
}};

void processAttributes(AtlasStruct entity, EntityMutationContext context, EntityMutations.EntityOperation operation) throws AtlasBaseException;
Expand Down
3 changes: 3 additions & 0 deletions .../main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/PreProcessorUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@ public class PreProcessorUtils {

public static final String DATA_PRODUCT_EDGE_LABEL = "__DataDomain.dataProducts";
public static final String DOMAIN_PARENT_EDGE_LABEL = "__DataDomain.subDomains";
public static final String STAKEHOLDER_EDGE_LABEL = "__DataDomain.stakeholders";
public static final String STAKEHOLDER_TITLE_EDGE_LABEL = "__StakeholderTitle.stakeholders";


public static final String PARENT_DOMAIN_QN_ATTR = "parentDomainQualifiedName";
public static final String SUPER_DOMAIN_QN_ATTR = "superDomainQualifiedName";
Expand Down
109 changes: 99 additions & 10 deletions .../apache/atlas/repository/store/graph/v2/preprocessor/datamesh/DataDomainPreProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,31 +18,35 @@
package org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh;


import com.sun.org.apache.bcel.internal.generic.NEW;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.RequestContext;
import org.apache.atlas.authorize.AtlasAuthorizationUtils;
import org.apache.atlas.authorize.AtlasEntityAccessRequest;
import org.apache.atlas.authorize.AtlasPrivilege;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasEntityHeader;
import org.apache.atlas.model.instance.AtlasObjectId;
import org.apache.atlas.model.instance.AtlasRelatedObjectId;
import org.apache.atlas.model.instance.AtlasStruct;
import org.apache.atlas.model.instance.EntityMutations;
import org.apache.atlas.model.instance.*;
import org.apache.atlas.repository.graph.GraphHelper;
import org.apache.atlas.repository.graphdb.AtlasGraph;
import org.apache.atlas.repository.graphdb.AtlasVertex;
import org.apache.atlas.repository.store.graph.AtlasEntityStore;
import org.apache.atlas.repository.store.graph.v2.AtlasGraphUtilsV2;
import org.apache.atlas.repository.store.graph.v2.EntityGraphRetriever;
import org.apache.atlas.repository.store.graph.v2.EntityMutationContext;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.atlas.utils.AtlasPerfMetrics;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

import static org.apache.atlas.repository.Constants.*;
import static org.apache.atlas.repository.graph.GraphHelper.getActiveChildrenVertices;
import static org.apache.atlas.repository.graph.GraphHelper.*;
import static org.apache.atlas.repository.store.graph.v2.preprocessor.PreProcessorUtils.*;
import static org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.StakeholderTitlePreProcessor.*;
import static org.apache.atlas.repository.util.AtlasEntityUtils.mapOf;

public class DataDomainPreProcessor extends AbstractDomainPreProcessor {
private static final Logger LOG = LoggerFactory.getLogger(DataDomainPreProcessor.class);
Expand All @@ -51,13 +55,15 @@ public class DataDomainPreProcessor extends AbstractDomainPreProcessor {
private Map<String, String> updatedPolicyResources;
private EntityGraphRetriever retrieverNoRelation = null;
private Map<String, String> updatedDomainQualifiedNames;
private AtlasEntityStore entityStore;

public DataDomainPreProcessor(AtlasTypeRegistry typeRegistry, EntityGraphRetriever entityRetriever,
AtlasGraph graph) {
AtlasGraph graph, AtlasEntityStore entityStore) {
super(typeRegistry, entityRetriever, graph);
this.updatedPolicyResources = new HashMap<>();
this.retrieverNoRelation = new EntityGraphRetriever(graph, typeRegistry, true);
this.updatedDomainQualifiedNames = new HashMap<>();
this.entityStore = entityStore;
}

@Override
Expand Down Expand Up @@ -112,6 +118,9 @@ private void processCreateDomain(AtlasEntity entity) throws AtlasBaseException {

entity.setAttribute(QUALIFIED_NAME, createQualifiedName(parentDomainQualifiedName));

// Check if authorized to create entities
AtlasAuthorizationUtils.verifyAccess(new AtlasEntityAccessRequest(typeRegistry, AtlasPrivilege.ENTITY_CREATE, new AtlasEntityHeader(entity)),
"create entity: type=", entity.getTypeName());

entity.setCustomAttributes(customAttributes);

Expand All @@ -131,6 +140,10 @@ private void processUpdateDomain(AtlasEntity entity, AtlasVertex vertex) throws
validateStakeholderRelationship(entity);

String vertexQnName = vertex.getProperty(QUALIFIED_NAME, String.class);
entity.setAttribute(QUALIFIED_NAME, vertexQnName);
// Check if authorized to update entities
AtlasAuthorizationUtils.verifyUpdateEntityAccess(typeRegistry, new AtlasEntityHeader(entity),"update entity: type=" + entity.getTypeName());


AtlasEntity storedDomain = entityRetriever.toAtlasEntity(vertex);
AtlasRelatedObjectId currentParentDomainObjectId = (AtlasRelatedObjectId) storedDomain.getRelationshipAttribute(PARENT_DOMAIN_REL_TYPE);
Expand Down Expand Up @@ -177,9 +190,8 @@ private void processUpdateDomain(AtlasEntity entity, AtlasVertex vertex) throws
if (!domainCurrentName.equals(domainNewName)) {
domainExists(domainNewName, currentParentDomainQualifiedName, storedDomain.getGuid());
}
entity.setAttribute(QUALIFIED_NAME, vertexQnName);
}

}
RequestContext.get().endMetricRecord(metricRecorder);
}

Expand Down Expand Up @@ -387,6 +399,83 @@ private void validateStakeholderRelationship(AtlasEntity entity) throws AtlasBas
throw new AtlasBaseException(AtlasErrorCode.OPERATION_NOT_SUPPORTED, "Managing Stakeholders while creating/updating a domain");
}
}

public List<AtlasEntityHeader> getStakeholderTitle(String domainQualifiedName) throws AtlasBaseException {
List<Map<String, Object>> mustClauseList = new ArrayList<>();
mustClauseList.add(mapOf("term", mapOf("__state", "ACTIVE")));
mustClauseList.add(mapOf("term", mapOf("__typeName.keyword", STAKEHOLDER_TITLE_ENTITY_TYPE)));

List<String> termsList = Arrays.asList(
NEW_STAR,
STAR,
domainQualifiedName
);

Map<String, Object> termsMap = mapOf(ATTR_DOMAIN_QUALIFIED_NAMES, termsList);
Map<String, Object> termsFilter = mapOf("terms", termsMap);

mustClauseList.add(termsFilter);

Map<String, Object> boolQuery = mapOf("must", mustClauseList);
Map<String, Object> query = mapOf("bool", boolQuery);
Map<String, Object> dsl = mapOf("query", query);

List<AtlasEntityHeader> assets = indexSearchPaginated(dsl, null, super.discovery);

return assets;
}


@Override
public void processDelete(AtlasVertex vertex) throws AtlasBaseException {
AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("processDomainDelete");

try{
List<String> stakeHolderGuids = new ArrayList<>();

// active childrens exists?
Iterator<AtlasVertex> childrens = getActiveChildrenVertices(vertex,
DOMAIN_PARENT_EDGE_LABEL, DATA_PRODUCT_EDGE_LABEL);
if (childrens.hasNext()){
throw new AtlasBaseException(AtlasErrorCode.OPERATION_NOT_SUPPORTED, "Domain cannot be archived because some subdomains or products are active in this domain");
}

// active stakeholder exists?
childrens = getActiveChildrenVertices(vertex, STAKEHOLDER_EDGE_LABEL);
while (childrens.hasNext()){
AtlasVertex child = childrens.next();
AtlasObjectId childId = entityRetriever.toAtlasObjectId(child);
stakeHolderGuids.add(childId.getGuid());
}

if (CollectionUtils.isNotEmpty(stakeHolderGuids)) {
entityStore.deleteByIds(stakeHolderGuids);
LOG.info("Deleted Stakeholders: {}", stakeHolderGuids);
}

// active stakeholder titles exists?
List<AtlasEntityHeader> stakeholderTitles = getStakeholderTitle(vertex.getProperty(QUALIFIED_NAME, String.class));
if (CollectionUtils.isNotEmpty(stakeholderTitles)) {
for (AtlasEntityHeader stakeholderTitle : stakeholderTitles) {
AtlasVertex stakeholderTitleVertex = entityRetriever.getEntityVertex(stakeholderTitle.getGuid());
AtlasGraphUtilsV2.removeItemFromListPropertyValue(stakeholderTitleVertex, ATTR_DOMAIN_QUALIFIED_NAMES, vertex.getProperty(QUALIFIED_NAME, String.class));
List<String> domainQualifiedNames = stakeholderTitleVertex.getMultiValuedProperty(ATTR_DOMAIN_QUALIFIED_NAMES, String.class);

if (CollectionUtils.isEmpty(domainQualifiedNames)) {
Iterator<AtlasVertex> stakeholders = getActiveChildrenVertices(stakeholderTitleVertex, STAKEHOLDER_TITLE_EDGE_LABEL);
if (!stakeholders.hasNext()) {
entityStore.deleteById(stakeholderTitle.getGuid());
LOG.info("Deleted Stakeholder Title: {}", stakeholderTitle.getGuid());
}
}
}
}

}
finally {
RequestContext.get().endMetricRecord(metricRecorder);
}
}
}


13 changes: 12 additions & 1 deletion ...apache/atlas/repository/store/graph/v2/preprocessor/datamesh/DataProductPreProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,9 @@
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.DeleteType;
import org.apache.atlas.RequestContext;
import org.apache.atlas.authorize.AtlasAuthorizationUtils;
import org.apache.atlas.authorize.AtlasEntityAccessRequest;
import org.apache.atlas.authorize.AtlasPrivilege;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.instance.*;
import org.apache.atlas.repository.graphdb.AtlasGraph;
Expand Down Expand Up @@ -99,6 +102,12 @@ private void processCreateProduct(AtlasEntity entity,AtlasVertex vertex) throws

entity.setAttribute(QUALIFIED_NAME, createQualifiedName(parentDomainQualifiedName));

// Check if authorized to create entities
AtlasAuthorizationUtils.verifyAccess(new AtlasEntityAccessRequest(typeRegistry, AtlasPrivilege.ENTITY_CREATE, new AtlasEntityHeader(entity)),
"create entity: type=", entity.getTypeName());

entity.setCustomAttributes(customAttributes);

productExists(productName, parentDomainQualifiedName, null);

createDaapVisibilityPolicy(entity, vertex);
Expand All @@ -114,6 +123,9 @@ private void processUpdateProduct(AtlasEntity entity, AtlasVertex vertex) throws
}

String vertexQnName = vertex.getProperty(QUALIFIED_NAME, String.class);
entity.setAttribute(QUALIFIED_NAME, vertexQnName);
// Check if authorized to update entities
AtlasAuthorizationUtils.verifyUpdateEntityAccess(typeRegistry, new AtlasEntityHeader(entity),"update entity: type=" + entity.getTypeName());

AtlasEntity storedProduct = entityRetriever.toAtlasEntity(vertex);
AtlasRelatedObjectId currentParentDomainObjectId = (AtlasRelatedObjectId) storedProduct.getRelationshipAttribute(DATA_DOMAIN_REL_TYPE);
Expand Down Expand Up @@ -161,7 +173,6 @@ private void processUpdateProduct(AtlasEntity entity, AtlasVertex vertex) throws
if (!productCurrentName.equals(productNewName)) {
productExists(productNewName, currentParentDomainQualifiedName, storedProduct.getGuid());
}
entity.setAttribute(QUALIFIED_NAME, vertexQnName);
}

if (isDaapVisibilityChanged) {
Expand Down
Loading
Loading